InfluxDB Bucket Retention Sentinel

id: influxdb-bucket-retention-sentinel
namespace: company.team
description: |
  Run Flux's buckets() function against an InfluxDB organization, flag any
  bucket with retentionPeriod == 0 (infinite retention - InfluxDB's own
  documented convention for "never expires"), alert Slack on breach, and
  conditionally bound that bucket's retention via the InfluxDB v2 REST
  API. auto_remediate and dry_run are two independent gates; no measurement
  data is ever deleted - only the retention window is shortened going
  forward.

triggers:
  - id: periodic_retention_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 6 hours. Shipped disabled; enable it once influxdb_org
      points at a real organization and a manual run has been verified. Always
      fires in safe mode (auto_remediate false, dry_run true) regardless of this
      flow's own defaults.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: influxdb_org
    type: STRING
    displayName: "InfluxDB Organization"
    defaults: "my-org"
    description: Organization whose buckets are audited, matching this plugin's own
      documented org property (see influxdb-load-query-data.yaml).

  - id: max_infinite_retention_buckets
    type: INT
    displayName: "Max Allowed Infinite-Retention Buckets"
    defaults: 0
    description: Breach when the count of buckets with retentionPeriod == 0 exceeds
      this number. Default 0 means any infinite-retention bucket at all is a
      breach.

  - id: remediation_retention_days
    type: INT
    displayName: "Remediation Retention (days)"
    defaults: 90
    description: Bounded retention applied to a flagged bucket when remediation
      runs, converted to InfluxDB's everySeconds field.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and infinite-retention buckets are found, PATCH each one
      to remediation_retention_days via the InfluxDB v2 REST API (still gated by
      dry_run below). If "false", the risk is only reported.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, no bucket is patched. Must
      be explicitly "false", together with auto_remediate "true", for the
      retention rule to actually change.

tasks:
  - id: audit_bucket_retention
    type: io.kestra.plugin.influxdb.FluxQuery
    description: Runs Flux's own buckets() function, which returns name, id,
      organizationID, retentionPolicy, and retentionPeriod (nanoseconds) for
      every bucket in influxdb_org - verified from Flux's documented stdlib.
      fetchType STORE is used because this plugin's own in-repo example
      (influxdb-load-query-data.yaml) documents the connection/org property
      pattern this task reuses.
    connection:
      url: "{{ secret('INFLUXDB_URL') }}"
      token: "{{ secret('INFLUXDB_TOKEN') }}"
    org: "{{ inputs.influxdb_org }}"
    query: "buckets()"
    fetchType: STORE

  - id: evaluate_retention_risk
    type: io.kestra.plugin.scripts.python.Script
    description: Reads the stored buckets() result and flags any bucket whose
      retentionPeriod is exactly 0 - InfluxDB's own documented value for
      infinite retention (confirmed from the InfluxDB v2 API's equivalent
      everySeconds field, "0 represents infinite retention"). The exact JSON
      shape this task's STORE output wraps the result in was not confirmed
      against a live instance, so the script handles both a bare row list and a
      dict-wrapped rows field defensively; see this blueprint's Pitfalls.
    inputFiles:
      buckets.json: "{{ outputs.audit_bucket_retention.uri }}"
    script: |
      import json
      from kestra import Kestra

      with open("buckets.json") as f:
          raw = json.load(f)

      rows = raw.get("rows", []) if isinstance(raw, dict) else raw

      infinite_buckets = [
          {"id": row.get("id"), "name": row.get("name")}
          for row in rows
          if row.get("retentionPeriod") == 0
      ]

      Kestra.outputs({
          "infinite_count": len(infinite_buckets),
          "infinite_buckets": infinite_buckets,
      })

  - id: evaluate_retention_gate
    type: io.kestra.plugin.core.flow.If
    description: Breach when at least one bucket has infinite retention.
    condition: "{{ outputs.evaluate_retention_risk.vars.infinite_count >
      inputs.max_infinite_retention_buckets }}"
    then:
      - id: route_remediation
        type: io.kestra.plugin.core.flow.Switch
        description: auto_remediate only decides whether remediation is considered at
          all; dry_run, checked separately below, decides whether the bucket is
          actually patched.
        value: "{{ inputs.auto_remediate }}"
        cases:
          "true":
            - id: check_dry_run
              type: io.kestra.plugin.core.flow.If
              condition: "{{ inputs.dry_run == \"false\" }}"
              then:
                - id: bound_bucket_retention
                  type: io.kestra.plugin.core.flow.Loop
                  description: Patch each flagged bucket's retention one at a time, so one failed
                    request never blocks the rest of the batch. There is no
                    structured bucket-management task in this plugin, so
                    remediation calls InfluxDB's own documented REST API
                    directly through the core Request task.
                  values: "{{ outputs.evaluate_retention_risk.vars.infinite_buckets }}"
                  concurrencyLimit: 4
                  tasks:
                    - id: patch_bucket_retention
                      type: io.kestra.plugin.core.http.Request
                      description: PATCH /api/v2/buckets/{bucketID}, InfluxDB's own documented
                        endpoint for updating retention rules, authenticated
                        with the Token scheme its own API reference specifies
                        (not Bearer).
                      uri: "{{ secret('INFLUXDB_URL') }}/api/v2/buckets/{{ item.value | jq('.id') |
                        first }}"
                      method: PATCH
                      contentType: application/json
                      headers:
                        Authorization: "Token {{ secret('INFLUXDB_TOKEN') }}"
                      body: |
                        {
                          "retentionRules": [
                            {
                              "type": "expire",
                              "everySeconds": {{ inputs.remediation_retention_days * 86400 }}
                            }
                          ]
                        }
                - id: log_remediation_applied
                  type: io.kestra.plugin.core.log.Log
                  message: "Bounded retention to {{ inputs.remediation_retention_days }} day(s) on
                    {{ outputs.evaluate_retention_risk.vars.infinite_count }}
                    bucket(s) in {{ inputs.influxdb_org }}."
              else:
                - id: log_dry_run_remediation
                  type: io.kestra.plugin.core.log.Log
                  message: "DRY RUN: would bound retention to {{ inputs.remediation_retention_days
                    }} day(s) on {{
                    outputs.evaluate_retention_risk.vars.infinite_count }}
                    bucket(s) in {{ inputs.influxdb_org }} (dry_run is true). No
                    change made."
          "false":
            - id: log_remediation_disabled
              type: io.kestra.plugin.core.log.Log
              message: "auto_remediate is false: {{
                outputs.evaluate_retention_risk.vars.infinite_count }} bucket(s)
                in {{ inputs.influxdb_org }} have infinite retention, but no
                patch was attempted."
      - id: alert_retention_risk
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Always alert on any breach, independent of whether it was remediated.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "InfluxDB bucket retention sentinel: {{ outputs.evaluate_retention_risk.vars.infinite_count }} bucket(s) in organization {{ inputs.influxdb_org }} have infinite retention (retentionPeriod == 0). auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ inputs.auto_remediate == \"true\" and inputs.dry_run == \"false\" }}. Execution {{ execution.id }}."
          }
    else:
      - id: log_retention_healthy
        type: io.kestra.plugin.core.log.Log
        message: "{{ inputs.influxdb_org }} is healthy: 0 buckets have infinite
          retention."

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Retention audit complete for {{ inputs.influxdb_org }}.
      infinite_count={{ outputs.evaluate_retention_risk.vars.infinite_count }}.
      threshold={{ inputs.max_infinite_retention_buckets }}. auto_remediate={{
      inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable InfluxDB instance or an invalid token.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "InfluxDB bucket retention sentinel FAILED in execution {{ execution.id }} for organization {{ inputs.influxdb_org }}. The retention audit did not complete; investigate before trusting the last known retention status."
      }