id: otel-collector-queue-loss-sentinel
namespace: company.team
description: Detect OpenTelemetry Collector exporter backpressure and failed
enqueue operations from Prometheus, with a fail-closed report and Slack alert.
inputs:
- id: prometheus_url
type: STRING
defaults: http://prometheus:9090
description: Prometheus server accessible from Kestra; uses the unauthenticated
HTTP API.
- id: occupancy_threshold
type: FLOAT
defaults: 0.8
description: Block when a queue reaches this fraction of capacity (0.01-1.0).
- id: max_failed_items
type: FLOAT
defaults: 0.0
description: Maximum allowed increase in failed enqueue items over the previous
10 minutes (>=0).
- id: max_age_seconds
type: INT
defaults: 600
description: Reject Prometheus vector samples older than this many seconds (60-3600).
triggers:
- id: collector_queue_check
type: io.kestra.plugin.core.trigger.Schedule
cron: "*/10 * * * *"
disabled: true
description: Shipped disabled until the Prometheus endpoint and exporter metrics
are verified.
tasks:
- id: queue_size
type: io.kestra.plugin.core.http.Request
description: Query queue size by Prometheus job, instance, exporter, and
collector instance ID.
uri: "{{ inputs.prometheus_url }}/api/v1/query"
method: GET
headers:
Accept: application/json
params:
query: 'sum by (job,instance,exporter,service_instance_id)
(otelcol_exporter_queue_size)'
- id: queue_capacity
type: io.kestra.plugin.core.http.Request
description: Query corresponding queue capacities; never treat missing capacity
as unlimited.
uri: "{{ inputs.prometheus_url }}/api/v1/query"
method: GET
headers:
Accept: application/json
params:
query: 'sum by (job,instance,exporter,service_instance_id)
(otelcol_exporter_queue_capacity)'
- id: enqueue_failures
type: io.kestra.plugin.core.http.Request
description: Sum increases across failed span, metric-point, and log-record
enqueue counters over ten minutes.
uri: "{{ inputs.prometheus_url }}/api/v1/query"
method: GET
headers:
Accept: application/json
params:
query: 'sum by (job,instance,exporter,service_instance_id)
(increase({__name__=~"otelcol_exporter_enqueue_failed_(spans|metric_points|log_records)"}[10m]))'
- id: assess_exporters
type: io.kestra.plugin.scripts.node.Script
description: Strictly align exporter identities, reject missing metrics and
stale samples, classify queue saturation or failed enqueue.
containerImage: node:22-alpine
env:
OCCUPANCY_THRESHOLD: "{{ inputs.occupancy_threshold }}"
MAX_FAILED_ITEMS: "{{ inputs.max_failed_items }}"
MAX_AGE_SECONDS: "{{ inputs.max_age_seconds }}"
inputFiles:
queue-size.json: "{{ outputs.queue_size.body | toJson }}"
queue-capacity.json: "{{ outputs.queue_capacity.body | toJson }}"
enqueue-failures.json: "{{ outputs.enqueue_failures.body | toJson }}"
outputFiles:
- collector-queue-report.json
script: |
const fs=require('fs');
function evaluateMetrics(sizeDoc,capDoc,failDoc,options,now){
const reasons=[],records=[];
function parse(doc,kind){
if(!doc||doc.status!=="success"||doc.data?.resultType!=="vector"||!Array.isArray(doc.data.result))throw Error("Invalid Prometheus "+kind+" query response");
if(doc.data.result.length>500)throw Error("Prometheus "+kind+" vector too large");
const values=new Map();
for(const row of doc.data.result){
if(!row||!row.metric||!Array.isArray(row.value)||row.value.length!==2)throw Error("Malformed "+kind+" metric");
const m=row.metric; if(typeof m.exporter!=="string"||!m.exporter)throw Error("Missing exporter label in "+kind);
const key=JSON.stringify([m.job??"",m.instance??"",m.exporter,m.service_instance_id??""]);
if(values.has(key))throw Error("Duplicate "+kind+" series for "+key);
const ts=Number(row.value[0]),raw=row.value[1],v=Number(raw);
if(raw===""||!Number.isFinite(v)||v<0)throw Error("Non-finite or negative "+kind+" sample");
if(!Number.isFinite(ts)||ts>now+60||now-ts>options.max_age_seconds)throw Error("Stale or future "+kind+" sample");
values.set(key,{labels:{job:m.job??"",instance:m.instance??"",exporter:m.exporter,service_instance_id:m.service_instance_id??""},value:v});
}
return values;
}
const sizes=parse(sizeDoc,"queue_size"),caps=parse(capDoc,"queue_capacity"),failures=parse(failDoc,"enqueue_failures");
if(sizes.size===0)reasons.push("No exporter queue_size metrics were returned");
for(const [key,size] of sizes){
const capacity=caps.get(key), failed=failures.get(key);
if(!capacity){reasons.push("Missing matching queue_capacity for "+key);continue}
if(!failed){reasons.push("Missing enqueue-failure counter for "+key);continue}
if(capacity.value<=0){reasons.push("Non-positive queue_capacity for "+key);continue}
if(size.value>capacity.value)reasons.push("Queue size exceeds capacity for "+key);
const ratio=size.value/capacity.value;
if(ratio>=options.occupancy_threshold)reasons.push("Queue occupancy above threshold for "+key);
if(failed.value>options.maximum_failed_items)reasons.push("Enqueue failures exceed limit for "+key);
records.push({...size.labels,queue_size:size.value,queue_capacity:capacity.value,occupancy:ratio,enqueue_failed_items_10m:failed.value});
}
for(const key of caps.keys())if(!sizes.has(key))reasons.push("queue_capacity without queue_size for "+key);
for(const key of failures.keys())if(!sizes.has(key))reasons.push("enqueue-failure metric without queue_size for "+key);
return {status:reasons.length?"BLOCK":"PASS",threshold:options.occupancy_threshold,max_enqueue_failed_items_10m:options.maximum_failed_items,exporters_checked:records.length,records,reasons};
}
function decode(name){
const v=JSON.parse(fs.readFileSync(name,'utf8'));
return typeof v==='string'?JSON.parse(v):v;
}
function setting(name,min,max){
const raw=process.env[name],n=Number(raw);
if(raw===undefined||raw===''||!Number.isFinite(n)||n<min||n>max)throw Error('Invalid '+name);
return n;
}
const options={
occupancy_threshold:setting('OCCUPANCY_THRESHOLD',0.01,1),
maximum_failed_items:setting('MAX_FAILED_ITEMS',0,1000000),
max_age_seconds:setting('MAX_AGE_SECONDS',60,3600)
};
const report=evaluateMetrics(decode('queue-size.json'),decode('queue-capacity.json'),decode('enqueue-failures.json'),options,Date.now()/1000);
fs.writeFileSync('collector-queue-report.json',JSON.stringify(report,null,2));
console.log('::'+JSON.stringify({outputs:{status:report.status,exporters_checked:report.exporters_checked,reason_count:report.reasons.length}})+'::');
console.log('OTel collector queue check: '+report.status+' exporters='+report.exporters_checked);
- id: report_location
type: io.kestra.plugin.core.output.OutputValues
description: Publish the verdict and saved report before the failure gate.
values:
status: "{{ outputs.assess_exporters.vars.status }}"
report_uri: "{{
outputs.assess_exporters.outputFiles['collector-queue-report.json'] }}"
exporters_checked: "{{ outputs.assess_exporters.vars.exporters_checked }}"
- id: pressure_gate
type: io.kestra.plugin.core.flow.If
description: Block the execution if a queue has high occupancy or failed enqueue
records.
condition: "{{ outputs.assess_exporters.vars.status == 'BLOCK' }}"
then:
- id: notify_exporter_pressure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Notify oncall with a safe numeric summary and report location.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{"text":"OpenTelemetry Collector queue preflight BLOCKED. Problems: {{ outputs.assess_exporters.vars.reason_count }}, matched exporters: {{ outputs.assess_exporters.vars.exporters_checked }}. Execution: {{ execution.id }}. Open the attached JSON report for per-exporter details."}
- id: stop_pressure_gate
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Collector exporter queues unhealthy or enqueue failures detected;
see collector-queue-report.json."
else:
- id: log_healthy
type: io.kestra.plugin.core.log.Log
message: "OpenTelemetry Collector queues PASS; {{
outputs.assess_exporters.vars.exporters_checked }} exporter queues
checked."
errors:
- id: log_failed_query
type: io.kestra.plugin.core.log.Log
message: "Collector queue monitor failed. Missing or invalid telemetry must
never be interpreted as healthy. Execution {{ execution.id }}."
