| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240 |
- 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
- }
|