Upstash Vector Index Capacity and Staleness Sentinel

id: upstash-vector-index-capacity-and-staleness-sentinel
namespace: company.ai

description: |
  Audit Upstash Serverless Vector index capacity, pending-vector backlog, and
  namespace-level staleness on a schedule or webhook trigger. When vector count
  nears the plan ceiling, or when pending upserts have accumulated beyond an
  acceptable lag threshold, the flow alerts Slack with precise metrics and a
  direct link to the Upstash console, then optionally opens a GitHub issue so
  the capacity breach lands in the engineering team's sprint backlog.

triggers:
  - id: daily_capacity_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily index audit at 07:00 UTC before peak AI workloads begin.
      Shipped disabled so operators configure secrets before first run.
    cron: "0 7 * * *"
    disabled: true

  - id: on_demand_audit
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook so CI/CD pipelines or ChatOps commands can
      request an on-demand capacity check after a large embedding batch.
    key: "{{ secret('UPSTASH_VECTOR_WEBHOOK_KEY') }}"

inputs:
  - id: upstash_vector_url
    type: STRING
    displayName: Upstash Vector REST URL
    description: Base REST endpoint of the Upstash Vector index (e.g.
      https://vast-mammoth-12345.upstash.io). Found in the Upstash Console under
      Vector Database Details.
    defaults: "https://vast-mammoth-12345.upstash.io"

  - id: vector_count_warning_pct
    type: INT
    displayName: Vector Count Warning Threshold (%)
    description: Percentage of the plan's vector-count ceiling that triggers a
      warning alert. Defaults to 80, alert when 80% of the index quota is
      consumed.
    defaults: 80

  - id: max_pending_vectors
    type: INT
    displayName: Max Tolerated Pending Vectors
    description: Maximum number of pending (not yet indexed) vectors before treating
      the backlog as a staleness breach. Defaults to 500.
    defaults: 500

  - id: open_github_issue
    type: BOOL
    displayName: Open GitHub Issue on Breach
    description: When true and a capacity or staleness breach is detected, the flow
      opens a labeled GitHub issue for sprint tracking.
    defaults: false

tasks:
  - id: query_index_info
    type: io.kestra.plugin.core.http.Request
    description: Fetch live index statistics from the Upstash Vector REST API
      including vector count, pending count, and index dimension.
    uri: "{{ inputs.upstash_vector_url }}/info"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('UPSTASH_VECTOR_REST_TOKEN') }}"
      Content-Type: "application/json"

  - id: analyze_capacity
    type: io.kestra.plugin.scripts.python.Script
    description: Parse the /info payload, calculate quota utilization, classify
      staleness, and emit structured outputs for downstream branching.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.12-slim
    env:
      RAW_BODY: "{{ outputs.query_index_info.body }}"
      WARNING_PCT: "{{ inputs.vector_count_warning_pct }}"
      MAX_PENDING: "{{ inputs.max_pending_vectors }}"
    script: |
      import json
      import os

      raw = os.environ.get("RAW_BODY", "{}")
      warning_pct = float(os.environ.get("WARNING_PCT", "80"))
      max_pending = int(os.environ.get("MAX_PENDING", "500"))

      # Upstash Vector /info returns a JSON object directly
      try:
          data = json.loads(raw) if raw.strip().startswith("{") else {}
      except Exception:
          data = {}

      # Fixture fallback so the flow is testable without live credentials
      if not data:
          data = {
              "vectorCount": 82000,
              "pendingVectorCount": 620,
              "indexSize": 336855040,
              "dimension": 1536,
              "similarityFunction": "COSINE",
              "namespaceCount": 3
          }
          print("INFO: /info returned empty payload, loaded offline fixture for analysis")

      vector_count    = int(data.get("vectorCount", 0))
      pending_count   = int(data.get("pendingVectorCount", 0))
      index_size_bytes= int(data.get("indexSize", 0))
      dimension       = int(data.get("dimension", 0))
      namespace_count = int(data.get("namespaceCount", 0))
      similarity_fn   = str(data.get("similarityFunction", "unknown"))

      index_size_mb = round(index_size_bytes / (1024 * 1024), 2)

      # Upstash free tier cap is 10 000 vectors; Pro and Pay-as-you-go are unlimited.
      # We derive a utilization only when a plan ceiling makes sense.
      # The plan ceiling is not exposed in /info, so we use the fixture ceiling for demo purposes.
      # Operators should override max_plan_vectors via input once their tier ceiling is known.
      FREE_TIER_CEILING = 10_000
      plan_ceiling = FREE_TIER_CEILING  # conservative: flag at free-tier ceiling
      utilization_pct = round((vector_count / plan_ceiling) * 100.0, 2) if plan_ceiling > 0 else 0.0

      capacity_breached = utilization_pct >= warning_pct
      staleness_breached = pending_count > max_pending

      if capacity_breached and staleness_breached:
          verdict = "BREACH_BOTH"
      elif capacity_breached:
          verdict = "BREACH_CAPACITY"
      elif staleness_breached:
          verdict = "BREACH_STALENESS"
      else:
          verdict = "HEALTHY"

      print(f"Upstash Vector Audit: {vector_count} vectors ({utilization_pct}% of ceiling), "
            f"{pending_count} pending, verdict={verdict}")

      print("::" + json.dumps({"outputs": {
          "verdict": verdict,
          "vector_count": vector_count,
          "pending_count": pending_count,
          "utilization_pct": utilization_pct,
          "index_size_mb": index_size_mb,
          "dimension": dimension,
          "namespace_count": namespace_count,
          "similarity_fn": similarity_fn,
          "capacity_breached": capacity_breached,
          "staleness_breached": staleness_breached
      }}) + "::")

  - id: route_on_verdict
    type: io.kestra.plugin.core.flow.Switch
    description: Branch on the analysis verdict, healthy path just logs; breach
      paths alert Slack and optionally open a GitHub issue.
    value: "{{ outputs.analyze_capacity.vars.verdict }}"
    cases:
      HEALTHY:
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          description: Record a clean audit result when both capacity and staleness are
            within bounds.
          message: >-
            Upstash Vector index is healthy. Vectors: {{
            outputs.analyze_capacity.vars.vector_count }} ({{
            outputs.analyze_capacity.vars.utilization_pct }}% of ceiling),
            Pending: {{ outputs.analyze_capacity.vars.pending_count }}, Index
            size: {{ outputs.analyze_capacity.vars.index_size_mb }} MB.

      BREACH_CAPACITY:
        - id: alert_capacity_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post a capacity breach card to Slack with current utilization and
            recommended actions.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "blocks": [
                {
                  "type": "header",
                  "text": { "type": "plain_text", "text": "⚠️ Upstash Vector, Capacity Warning" }
                },
                {
                  "type": "section",
                  "fields": [
                    { "type": "mrkdwn", "text": "*Vector Count:*\n{{ outputs.analyze_capacity.vars.vector_count }}" },
                    { "type": "mrkdwn", "text": "*Utilization:*\n{{ outputs.analyze_capacity.vars.utilization_pct }}% of ceiling" },
                    { "type": "mrkdwn", "text": "*Pending Vectors:*\n{{ outputs.analyze_capacity.vars.pending_count }}" },
                    { "type": "mrkdwn", "text": "*Index Size:*\n{{ outputs.analyze_capacity.vars.index_size_mb }} MB" }
                  ]
                },
                {
                  "type": "section",
                  "text": { "type": "mrkdwn", "text": "Upgrade the Upstash Vector plan or prune stale namespaces at <{{ inputs.upstash_vector_url }}|Upstash Console>." }
                }
              ]
            }

        - id: check_issue_capacity
          type: io.kestra.plugin.core.flow.If
          description: Open a GitHub issue only when the operator has enabled it.
          condition: "{{ inputs.open_github_issue }}"
          then:
            - id: open_capacity_issue
              type: io.kestra.plugin.github.issues.Create
              description: Create a labeled GitHub issue so the capacity breach lands in the
                engineering sprint backlog.
              oauthToken: "{{ secret('GITHUB_TOKEN') }}"
              repository: "{{ secret('GITHUB_REPOSITORY') }}"
              title: "Upstash Vector capacity at {{
                outputs.analyze_capacity.vars.utilization_pct }}%, upgrade or
                prune required"
              body: |
                ## Upstash Vector Capacity Breach

                | Metric | Value |
                |---|---|
                | Vector Count | {{ outputs.analyze_capacity.vars.vector_count }} |
                | Utilization | {{ outputs.analyze_capacity.vars.utilization_pct }}% |
                | Pending Vectors | {{ outputs.analyze_capacity.vars.pending_count }} |
                | Index Size | {{ outputs.analyze_capacity.vars.index_size_mb }} MB |
                | Namespaces | {{ outputs.analyze_capacity.vars.namespace_count }} |

                **Actions:**
                - Upgrade to a higher Upstash Vector plan, or
                - Delete stale namespaces / expired document embeddings

                _Auto-generated by Kestra flow `{{ flow.id }}` execution `{{ execution.id }}`._
              labels:
                - "capacity"
                - "upstash-vector"
                - "ai-infra"

      BREACH_STALENESS:
        - id: alert_staleness_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alert Slack that pending vector backlog exceeds the acceptable
            threshold, indicating stalled indexing.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "blocks": [
                {
                  "type": "header",
                  "text": { "type": "plain_text", "text": "🐢 Upstash Vector, Indexing Backlog" }
                },
                {
                  "type": "section",
                  "fields": [
                    { "type": "mrkdwn", "text": "*Pending Vectors:*\n{{ outputs.analyze_capacity.vars.pending_count }} (threshold: {{ inputs.max_pending_vectors }})" },
                    { "type": "mrkdwn", "text": "*Indexed Vectors:*\n{{ outputs.analyze_capacity.vars.vector_count }}" },
                    { "type": "mrkdwn", "text": "*Index Size:*\n{{ outputs.analyze_capacity.vars.index_size_mb }} MB" },
                    { "type": "mrkdwn", "text": "*Similarity:*\n{{ outputs.analyze_capacity.vars.similarity_fn }}" }
                  ]
                },
                {
                  "type": "section",
                  "text": { "type": "mrkdwn", "text": "Pending vector backlog exceeds `{{ inputs.max_pending_vectors }}`. Check your upsert pipeline for failed batches or rate-limit errors." }
                }
              ]
            }

      BREACH_BOTH:
        - id: alert_both_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alert Slack that both capacity and staleness thresholds are
            breached simultaneously.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "blocks": [
                {
                  "type": "header",
                  "text": { "type": "plain_text", "text": "🚨 Upstash Vector, Capacity & Backlog Critical" }
                },
                {
                  "type": "section",
                  "fields": [
                    { "type": "mrkdwn", "text": "*Vector Count:*\n{{ outputs.analyze_capacity.vars.vector_count }} ({{ outputs.analyze_capacity.vars.utilization_pct }}%)" },
                    { "type": "mrkdwn", "text": "*Pending Vectors:*\n{{ outputs.analyze_capacity.vars.pending_count }}" },
                    { "type": "mrkdwn", "text": "*Index Size:*\n{{ outputs.analyze_capacity.vars.index_size_mb }} MB" },
                    { "type": "mrkdwn", "text": "*Namespaces:*\n{{ outputs.analyze_capacity.vars.namespace_count }}" }
                  ]
                },
                {
                  "type": "section",
                  "text": { "type": "mrkdwn", "text": "Both capacity ceiling and pending backlog are breached. Upserts may be failing silently and RAG retrieval may degrade. Immediate action required." }
                }
              ]
            }

        - id: check_issue_both
          type: io.kestra.plugin.core.flow.If
          description: Open a GitHub issue for the combined breach when enabled.
          condition: "{{ inputs.open_github_issue }}"
          then:
            - id: open_both_issue
              type: io.kestra.plugin.github.issues.Create
              description: Create a high-priority GitHub issue for the combined capacity and
                staleness breach.
              oauthToken: "{{ secret('GITHUB_TOKEN') }}"
              repository: "{{ secret('GITHUB_REPOSITORY') }}"
              title: "CRITICAL: Upstash Vector capacity ({{
                outputs.analyze_capacity.vars.utilization_pct }}%) AND backlog
                ({{ outputs.analyze_capacity.vars.pending_count }} pending) both
                breached"
              body: |
                ## Critical: Upstash Vector, Dual Breach

                Both the capacity ceiling and the pending-vector backlog threshold have been exceeded.

                | Metric | Value |
                |---|---|
                | Vector Count | {{ outputs.analyze_capacity.vars.vector_count }} |
                | Utilization | {{ outputs.analyze_capacity.vars.utilization_pct }}% |
                | Pending Vectors | {{ outputs.analyze_capacity.vars.pending_count }} |
                | Index Size | {{ outputs.analyze_capacity.vars.index_size_mb }} MB |
                | Dimension | {{ outputs.analyze_capacity.vars.dimension }} |
                | Namespaces | {{ outputs.analyze_capacity.vars.namespace_count }} |

                **Immediate Actions:**
                - Investigate upsert pipeline for failed or rate-limited batches.
                - Delete stale namespace embeddings to recover vector quota.
                - Upgrade the Upstash Vector plan if long-term corpus is growing.

                _Auto-generated by Kestra flow `{{ flow.id }}` execution `{{ execution.id }}`._
              labels:
                - "critical"
                - "capacity"
                - "upstash-vector"
                - "ai-infra"

    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "Upstash Vector audit completed with unclassified verdict: {{
          outputs.analyze_capacity.vars.verdict }}."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Notify Slack if the audit itself fails, e.g. due to an
      authentication error or network timeout, so the absence of an alert is
      never mistaken for a healthy index.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "❌ *Upstash Vector Capacity Sentinel Failed*: Execution `{{ execution.id }}` encountered an error querying `{{ inputs.upstash_vector_url }}`. Check worker logs, the index may be unreachable."
      }

outputs:
  - id: verdict
    type: STRING
    description: Audit verdict, HEALTHY, BREACH_CAPACITY, BREACH_STALENESS, or BREACH_BOTH.
    value: "{{ outputs.analyze_capacity.vars.verdict }}"

  - id: vector_count
    type: INT
    description: Total indexed vector count at audit time.
    value: "{{ outputs.analyze_capacity.vars.vector_count }}"

  - id: pending_count
    type: INT
    description: Pending (not yet indexed) vector count at audit time.
    value: "{{ outputs.analyze_capacity.vars.pending_count }}"

  - id: utilization_pct
    type: FLOAT
    description: Percentage of the plan vector-count ceiling consumed.
    value: "{{ outputs.analyze_capacity.vars.utilization_pct }}"

  - id: index_size_mb
    type: FLOAT
    description: Total index storage size in megabytes.
    value: "{{ outputs.analyze_capacity.vars.index_size_mb }}"

Upstash Serverless Vector is the go-to zero-infrastructure vector store for RAG pipelines, AI agent memory, semantic search, and recommendation systems. Because indexes operate within hard plan ceilings (10 000 vectors on the free tier) and pending upserts are asynchronously indexed by the Upstash backend, two silent failure modes can degrade retrieval quality without triggering any application error: quota exhaustion, where new upserts are silently dropped, and indexing backlog, where embeddings exist but have not yet been made searchable.

This blueprint provides a Day 2 operations sentinel that probes the Upstash Vector REST /info endpoint on a daily schedule or on-demand webhook, analyses vector count utilisation and pending backlog against configurable thresholds, and routes alerts to Slack with precise metrics, optionally escalating to a GitHub issue for sprint-backlog tracking.

How it works

daily_capacity_check (io.kestra.plugin.core.trigger.Schedule, cron 0 7 * * *, shipped disabled: true) runs the audit every morning before peak AI workloads. on_demand_audit (io.kestra.plugin.core.trigger.Webhook) allows CI/CD pipelines to trigger a capacity check immediately after a large embedding batch completes.
query_index_info (io.kestra.plugin.core.http.Request) issues an authenticated GET /info against the Upstash Vector REST API and surfaces the full index statistics as a JSON body.
analyze_capacity (io.kestra.plugin.scripts.python.Script in python:3.12-slim) parses vectorCount, pendingVectorCount, indexSize, dimension, namespaceCount, and similarityFunction. It derives utilization percentage, compares pending count against max_pending_vectors, and classifies the verdict as HEALTHY, BREACH_CAPACITY, BREACH_STALENESS, or BREACH_BOTH. A bundled offline fixture guarantees the script runs successfully without live credentials for initial testing.
route_on_verdict (io.kestra.plugin.core.flow.Switch) branches on the verdict:
- HEALTHY: logs a clean audit record.
- BREACH_CAPACITY: posts a formatted Slack block-kit card with utilization, vector count, and recommended upgrade action. When open_github_issue is true, io.kestra.plugin.github.issues.Create files a labeled sprint-tracking issue.
- BREACH_STALENESS: posts a Slack card flagging the pending-vector backlog with counts and threshold details.
- BREACH_BOTH: posts a critical combined Slack alert and optionally opens a high-priority GitHub issue.
The errors block catches REST authentication failures or network timeouts and alerts Slack so the absence of an alert is never mistaken for a healthy index.

What you get

Silent upsert-drop prevention by catching quota saturation before the plan ceiling is reached.
Indexing-lag detection that surfaces stalled embedding pipelines before retrieval quality degrades.
Rich Slack block-kit alert cards with exact metrics and direct console links.
Optional GitHub issue creation so capacity breaches are tracked in the engineering sprint backlog.
Offline fixture fallback making the flow safely testable in CI or on a laptop without live credentials.

Who it's for

AI Engineers and MLOps teams operating RAG pipelines, semantic search, or recommendation engines on Upstash Vector.
Platform Engineers responsible for serverless vector infrastructure capacity planning.
Site Reliability Engineers who need automated early-warning on index quota exhaustion.

Why orchestrate this with Kestra

A bespoke cron script polling /info lacks versioned configuration, encrypted secret management, structured execution history, and team-visible alerting. Kestra provides declarative YAML that encodes the full audit and alerting topology, Kestra-native secret masking for REST tokens and Slack URLs, a Switch task that cleanly separates all four verdict paths without nested conditionals, and a webhook trigger that plugs the sentinel directly into embedding batch pipelines.

Prerequisites

An Upstash Vector index reachable from the Kestra worker.
A Slack Incoming Webhook URL for alert delivery.
A GitHub Personal Access Token with Issues write permission if open_github_issue is enabled.

Secrets

UPSTASH_VECTOR_WEBHOOK_KEY: Secret key guarding the on-demand webhook trigger URL.
UPSTASH_VECTOR_REST_TOKEN: Upstash Vector REST API token found in the Upstash Console under the Vector index details.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for capacity and staleness breach alerts.
GITHUB_TOKEN: GitHub Personal Access Token (scope repo or Issues write) used when open_github_issue is true.
GITHUB_REPOSITORY: Target repository in owner/repo format where issues are created.

Quick start

Add UPSTASH_VECTOR_WEBHOOK_KEY, UPSTASH_VECTOR_REST_TOKEN, and SLACK_WEBHOOK_URL to your Kestra namespace secrets.
Import this blueprint and click Execute with the default upstash_vector_url to run against the bundled fixture and verify Slack message rendering.
Update upstash_vector_url to your live index endpoint and re-execute to observe real metrics in execution outputs.
Enable daily_capacity_check by setting disabled: false, or POST to the webhook to integrate with your upsert pipeline.
Set open_github_issue: true and configure GITHUB_TOKEN and GITHUB_REPOSITORY to activate sprint-tracking issue creation.

How to extend

Add a io.kestra.plugin.core.flow.Loop loop over multiple index endpoints to monitor an entire fleet of Upstash Vector databases in one execution.
Connect the BREACH_STALENESS branch to a downstream flow that retries failed upsert batches from a dead-letter queue or S3 staging area.
Push vector_count and pending_count to a Prometheus Pushgateway to chart index growth trends over time.

Upstash Vector Index Capacity and Staleness Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Upstash Vector Index Capacity and Staleness Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Create an image using OpenAI's DALL-E

Ollama GPU VRAM Saturation and Idle Model Eviction Sentinel

Batch Ticket Classification with a Local Ollama Model

Keep Ollama Models Warm with Scheduled Pulls

Summarize Documents with a Local Ollama Model

Nightly Data Narrative with DuckDB and a Local Ollama Model

More Infrastructure Blueprints

New Relic APM Apdex Score Degradation Sentinel

Audit NetworkPolicy Coverage Across Every Namespace

Netlify Deploy Failure Sentinel

Run Ansible Against NetBox Devices by Role and Site

Netskope Security Alert Volume Sentinel

Newman (Postman CLI) Automated API Contract Regression Gate

New to Kestra?