package shared import "fmt" // Health represents per-component pipeline health. type Health int8 const ( HealthOK Health = iota // records flowing, no errors HealthWarning // retries, throughput anomaly HealthError // errors, dropped records, retries_failed HealthIdle // zero throughput — possible misconfig ) func (h Health) String() string { return [...]string{"ok", "warn", "error", "idle"}[h] } // Badge returns a short colored indicator for table display. func (h Health) Badge() string { switch h { case HealthOK: return "●" case HealthWarning: return "▲" case HealthError: return "✖" case HealthIdle: return "○" default: return "?" } } // Diag is a single diagnostic message for the warnings panel. type Diag struct { Level Health CompID string // component ID, or "" for global Kind Kind // component kind Message string Hint string // suggested config area to check } // AnalyzeDiagnostics inspects a State and produces actionable diagnostics. func AnalyzeDiagnostics(s State, prevComps []ComponentMetrics) []Diag { var diags []Diag prevMap := make(map[string]ComponentMetrics, len(prevComps)) for _, c := range prevComps { prevMap[c.ID] = c } for _, c := range s.Comps { diags = append(diags, componentDiags(c, prevMap[c.ID])...) } // Global diagnostics totalOut, totalIn := flowTotals(s.Comps) if totalIn > 0 && totalOut == 0 { diags = append(diags, Diag{ Level: HealthError, Message: "No output records — pipeline is blocked or outputs misconfigured", Hint: "Check output plugin destination, auth, and network connectivity", }) } return diags } // ComponentHealth returns the health status for a single component. func ComponentHealth(c ComponentMetrics, prev ComponentMetrics) Health { switch c.Kind { case KindOutput: if c.Fields["errors"] > 0 || c.Fields["retries_failed"] > 0 { return HealthError } if c.Fields["retries"] > 0 || c.Fields["dropped_records"] > 0 { return HealthWarning } if c.Records() == 0 { return HealthIdle } case KindFilter: total := c.Records() + c.Fields["drop_records"] if total > 0 { dropRatio := float64(c.Fields["drop_records"]) / float64(total) if dropRatio > 0.5 { return HealthWarning } } if c.Records() == 0 { return HealthIdle } case KindInput: if c.Records() == 0 { return HealthIdle } } return HealthOK } // PipelineSummary is a high-level summary for the header. type PipelineSummary struct { TotalInputs int TotalFilters int TotalOutputs int TotalErrors int64 TotalRetries int64 TotalDropped int64 InputRec int64 OutputRec int64 HasErrors bool HasWarnings bool Flowing bool } func SummarizePipeline(comps []ComponentMetrics) PipelineSummary { var ps PipelineSummary for _, c := range comps { switch c.Kind { case KindInput: ps.TotalInputs++ ps.InputRec += c.Records() case KindFilter: ps.TotalFilters++ case KindOutput: ps.TotalOutputs++ ps.OutputRec += c.Records() ps.TotalErrors += c.Fields["errors"] ps.TotalRetries += c.Fields["retries"] ps.TotalDropped += c.Fields["dropped_records"] } } ps.HasErrors = ps.TotalErrors > 0 ps.HasWarnings = ps.TotalRetries > 0 || ps.TotalDropped > 0 ps.Flowing = ps.InputRec > 0 && ps.OutputRec > 0 return ps } func componentDiags(c, prev ComponentMetrics) []Diag { var diags []Diag h := ComponentHealth(c, prev) switch h { case HealthError: if c.Kind == KindOutput { if e := c.Fields["errors"]; e > 0 { diags = append(diags, Diag{ Level: HealthError, CompID: c.ID, Kind: c.Kind, Message: fmt.Sprintf("%d processing errors", e), Hint: "Check output destination availability, TLS certs, and credentials", }) } if rf := c.Fields["retries_failed"]; rf > 0 { diags = append(diags, Diag{ Level: HealthError, CompID: c.ID, Kind: c.Kind, Message: fmt.Sprintf("%d failed retries", rf), Hint: "Destination may be unreachable — check network, DNS, and firewall rules", }) } } case HealthWarning: if c.Kind == KindOutput { if r := c.Fields["retries"]; r > 0 { diags = append(diags, Diag{ Level: HealthWarning, CompID: c.ID, Kind: c.Kind, Message: fmt.Sprintf("%d retries", r), Hint: "Transient failures — monitor or increase retry_limit in output config", }) } if d := c.Fields["dropped_records"]; d > 0 { diags = append(diags, Diag{ Level: HealthWarning, CompID: c.ID, Kind: c.Kind, Message: fmt.Sprintf("%d records dropped", d), Hint: "Check queue_limit / mem_buf_limit in output config", }) } } if c.Kind == KindFilter { total := c.Records() + c.Fields["drop_records"] if total > 0 && float64(c.Fields["drop_records"])/float64(total) > 0.5 { diags = append(diags, Diag{ Level: HealthWarning, CompID: c.ID, Kind: c.Kind, Message: fmt.Sprintf("dropping >%d%% of records", int(float64(c.Fields["drop_records"])/float64(total)*100)), Hint: "Verify filter conditions — high drop rate may indicate misconfigured rules", }) } } case HealthIdle: switch c.Kind { case KindInput: diags = append(diags, Diag{ Level: HealthIdle, CompID: c.ID, Kind: c.Kind, Message: "No records collected", Hint: "Check path/tag/source in input config — file may not exist or be empty", }) case KindOutput: diags = append(diags, Diag{ Level: HealthIdle, CompID: c.ID, Kind: c.Kind, Message: "No records processed", Hint: "Upstream may be blocked — verify filters are not dropping all records", }) case KindFilter: diags = append(diags, Diag{ Level: HealthIdle, CompID: c.ID, Kind: c.Kind, Message: "No records passed through", Hint: "Check filter match pattern and upstream output tags", }) } } return diags } func flowTotals(comps []ComponentMetrics) (in, out int64) { for _, c := range comps { switch c.Kind { case KindInput: in += c.Records() case KindOutput: out += c.Records() } } return }