| 1 | package framework |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi" |
| 7 | "strings" |
| 8 | ) |
| 9 | |
| 10 | // Collector is the base type for all framework-based collectors |
| 11 | type Collector struct { |
| 12 | collectorapi.Base |
| 13 | |
| 14 | Config Config |
| 15 | State *CollectorState |
| 16 | registeredContexts []any // All contexts from generated code |
| 17 | contextMap map[string]any |
| 18 | charts *collectorapi.Charts |
| 19 | impl CollectorImpl // The actual collector implementation |
| 20 | globalLabels []collectorapi.Label // Job-level labels applied to all charts |
| 21 | instanceCharts map[string]struct{} |
| 22 | } |
| 23 | |
| 24 | // Init initializes the collector (go.d framework requirement) |
| 25 | func (c *Collector) Init(ctx context.Context) error { |
| 26 | // Initialize state |
| 27 | c.State = NewCollectorState() |
| 28 | c.State.collector = &c.Base // Set logger reference |
| 29 | c.contextMap = make(map[string]any) |
| 30 | c.charts = &collectorapi.Charts{} |
| 31 | c.globalLabels = make([]collectorapi.Label, 0) |
| 32 | c.instanceCharts = make(map[string]struct{}) |
| 33 | |
| 34 | // Set defaults |
| 35 | if c.Config.ObsoletionIterations == 0 { |
| 36 | c.Config.ObsoletionIterations = 60 |
| 37 | } |
| 38 | if c.Config.UpdateEvery == 0 { |
| 39 | c.Config.UpdateEvery = 1 |
| 40 | } |
| 41 | |
| 42 | // Validate configuration |
| 43 | if err := c.validateConfig(); err != nil { |
| 44 | return fmt.Errorf("invalid configuration: %v", err) |
| 45 | } |
| 46 | |
| 47 | return nil |
| 48 | } |
| 49 | |
| 50 | // Check tests connectivity (go.d framework requirement) |
| 51 | func (c *Collector) Check(ctx context.Context) error { |
| 52 | // This will be overridden by specific collectors |
| 53 | // to test protocol connectivity |
| 54 | return nil |
| 55 | } |
| 56 | |
| 57 | // Charts returns the chart definitions (go.d framework requirement) |
| 58 | func (c *Collector) Charts() *collectorapi.Charts { |
| 59 | return c.charts |
| 60 | } |
| 61 | |
| 62 | // Collect gathers metrics (go.d framework requirement) |
| 63 | func (c *Collector) Collect(ctx context.Context) map[string]int64 { |
| 64 | // Increment global iteration counter |
| 65 | c.State.iteration++ |
| 66 | |
| 67 | // Clear previous iteration errors |
| 68 | c.State.ClearErrors() |
| 69 | |
| 70 | // Run collection (implemented by specific collector) |
| 71 | var err error |
| 72 | if c.impl != nil { |
| 73 | err = c.impl.CollectOnce() |
| 74 | } else { |
| 75 | err = fmt.Errorf("collector implementation not set") |
| 76 | } |
| 77 | |
| 78 | // Handle errors (let the module decide how to handle them) |
| 79 | if err != nil { |
| 80 | c.Errorf("collection failed: %v", err) |
| 81 | // Return whatever metrics we have (partial collection is OK) |
| 82 | } |
| 83 | |
| 84 | // Convert collected metrics to go.d format |
| 85 | metrics := c.convertMetrics() |
| 86 | |
| 87 | // Advance iteration and handle obsoletion |
| 88 | c.State.NextIteration(c.Config.ObsoletionIterations) |
| 89 | |
| 90 | // Handle obsolete instances - mark their charts as obsolete |
| 91 | for _, instanceKey := range c.State.GetObsoleteInstances() { |
| 92 | c.markChartsObsolete(instanceKey) |
| 93 | } |
| 94 | |
| 95 | return metrics |
| 96 | } |
| 97 | |
| 98 | // Cleanup performs cleanup (go.d framework requirement) |
| 99 | func (c *Collector) Cleanup(ctx context.Context) { |
| 100 | // This will be overridden by specific collectors |
| 101 | // to close connections, etc. |
| 102 | } |
| 103 | |
| 104 | // SetImpl sets the collector implementation |
| 105 | func (c *Collector) SetImpl(impl CollectorImpl) { |
| 106 | c.impl = impl |
| 107 | } |
| 108 | |
| 109 | // validateConfig validates the configuration |
| 110 | func (c *Collector) validateConfig() error { |
| 111 | // Validate update intervals are multiples of base |
| 112 | for groupName, interval := range c.Config.CollectionGroups { |
| 113 | if interval%c.Config.UpdateEvery != 0 { |
| 114 | // Adjust to nearest valid multiple |
| 115 | adjusted := ((interval / c.Config.UpdateEvery) + 1) * c.Config.UpdateEvery |
| 116 | c.Config.CollectionGroups[groupName] = adjusted |
| 117 | c.Infof("adjusted %s interval from %d to %d (must be multiple of %d)", |
| 118 | groupName, interval, adjusted, c.Config.UpdateEvery) |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | return nil |
| 123 | } |
| 124 | |
| 125 | // convertMetrics converts framework metrics to go.d format |
| 126 | func (c *Collector) convertMetrics() map[string]int64 { |
| 127 | mx := make(map[string]int64) |
| 128 | |
| 129 | // Track seen instances for dynamic chart creation |
| 130 | seenInstances := make(map[string]bool) |
| 131 | |
| 132 | // Convert instance metrics |
| 133 | for _, metric := range c.State.GetMetrics() { |
| 134 | // Track this instance |
| 135 | instanceKey := metric.Instance.key |
| 136 | seenInstances[instanceKey] = true |
| 137 | |
| 138 | // Check if we need to create charts for this instance |
| 139 | if !c.hasChartsForInstance(instanceKey) { |
| 140 | // Find the context this instance belongs to |
| 141 | for _, ctx := range c.registeredContexts { |
| 142 | contextMeta := extractContextMetadata(ctx) |
| 143 | if contextMeta != nil && contextMeta.Name == metric.Instance.contextName { |
| 144 | // Get the current instance from state (not the copy in metric) |
| 145 | currentInstance := c.State.instances[instanceKey] |
| 146 | if currentInstance == nil { |
| 147 | // Fallback to metric instance if not found (shouldn't happen) |
| 148 | currentInstance = &metric.Instance |
| 149 | } |
| 150 | // Create charts for this new instance |
| 151 | chart := c.createChartFromContext(ctx, instanceKey, currentInstance) |
| 152 | if chart != nil { |
| 153 | c.Debugf("Creating dynamic chart for instance: %s", instanceKey) |
| 154 | // Add labels from the instance |
| 155 | for k, v := range metric.Instance.labels { |
| 156 | chart.Labels = append(chart.Labels, collectorapi.Label{ |
| 157 | Key: k, |
| 158 | Value: v, |
| 159 | }) |
| 160 | } |
| 161 | if c.charts.Has(chart.ID) { |
| 162 | c.markInstanceChartsPresent(instanceKey) |
| 163 | continue |
| 164 | } |
| 165 | if err := c.charts.Add(chart); err != nil { |
| 166 | c.Errorf("failed adding chart for %s: %v", instanceKey, err) |
| 167 | } else { |
| 168 | c.markInstanceChartsPresent(instanceKey) |
| 169 | } |
| 170 | } |
| 171 | break |
| 172 | } |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | // Generate go.d metric key |
| 177 | key := c.generateMetricKey(metric) |
| 178 | mx[key] = metric.Value |
| 179 | } |
| 180 | |
| 181 | // Add protocol observability metrics (if any protocols are registered) |
| 182 | // This will be added when protocol observability charts are created |
| 183 | // for name, pm := range c.State.protocols { |
| 184 | // prefix := fmt.Sprintf("protocol_%s_", name) |
| 185 | // mx[prefix+"requests"] = pm.RequestCount |
| 186 | // mx[prefix+"errors"] = pm.ErrorCount |
| 187 | // if pm.RequestCount > 0 { |
| 188 | // mx[prefix+"avg_latency"] = pm.TotalLatency / pm.RequestCount |
| 189 | // } |
| 190 | // mx[prefix+"max_latency"] = pm.MaxLatency |
| 191 | // mx[prefix+"bytes_sent"] = pm.BytesSent |
| 192 | // mx[prefix+"bytes_received"] = pm.BytesReceived |
| 193 | // } |
| 194 | |
| 195 | return mx |
| 196 | } |
| 197 | |
| 198 | // generateMetricKey creates the go.d metric key from instance and dimension |
| 199 | func (c *Collector) generateMetricKey(metric MetricValue) string { |
| 200 | // Use our new scheme: {instance_id}.{dimension_name} |
| 201 | return metric.Instance.key + "." + metric.Dimension |
| 202 | } |
| 203 | |
| 204 | // cleanLabelValue cleans a label value for use in metric keys |
| 205 | func cleanLabelValue(value string) string { |
| 206 | // Replace problematic characters |
| 207 | r := strings.NewReplacer( |
| 208 | " ", "_", |
| 209 | ".", "_", |
| 210 | "-", "_", |
| 211 | "/", "_", |
| 212 | ":", "_", |
| 213 | "=", "_", |
| 214 | ) |
| 215 | return strings.ToLower(r.Replace(value)) |
| 216 | } |
| 217 | |
| 218 | // createChartFromContext creates a go.d chart from a Context[T] |
| 219 | func (c *Collector) createChartFromContext(ctx any, instanceID string, instance *Instance) *collectorapi.Chart { |
| 220 | // Use reflection to extract context metadata |
| 221 | contextMeta := extractContextMetadata(ctx) |
| 222 | if contextMeta == nil { |
| 223 | return nil |
| 224 | } |
| 225 | |
| 226 | // Create chart with unique ID for dynamic instances |
| 227 | // For labeled contexts, include the instance ID in the chart ID |
| 228 | chartID := cleanChartID(instanceID) |
| 229 | if !contextMeta.HasLabels { |
| 230 | // For unlabeled contexts, use just the context name |
| 231 | chartID = cleanChartID(contextMeta.Name) |
| 232 | } |
| 233 | |
| 234 | chart := &collectorapi.Chart{ |
| 235 | ID: chartID, |
| 236 | // OverID: instanceID, // Commented out - let go.d framework handle chart naming |
| 237 | Title: contextMeta.Title, |
| 238 | Units: contextMeta.Units, |
| 239 | Fam: contextMeta.Family, |
| 240 | Ctx: contextMeta.Name, |
| 241 | Type: contextMeta.Type, |
| 242 | Priority: contextMeta.Priority, |
| 243 | Opts: collectorapi.Opts{}, |
| 244 | } |
| 245 | |
| 246 | // Set UpdateEvery override if instance has one |
| 247 | if instance != nil && instance.UpdateEveryOverride > 0 { |
| 248 | chart.UpdateEvery = instance.UpdateEveryOverride |
| 249 | // For charts with longer update intervals (like statistics), skip gaps |
| 250 | if instance.UpdateEveryOverride > 1 { |
| 251 | chart.SkipGaps = true |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | // Add dimensions with full dimension IDs |
| 256 | instancePrefix := instanceID + "." |
| 257 | for _, dim := range contextMeta.Dimensions { |
| 258 | // Use the full dimension ID: {instance_id}.{dimension_name} |
| 259 | dimID := instancePrefix + dim.Name |
| 260 | chartDim := &collectorapi.Dim{ |
| 261 | ID: dimID, |
| 262 | Name: dim.Name, |
| 263 | Algo: dim.Algorithm, |
| 264 | Mul: 1, // Always 1 - values are already in base units |
| 265 | Div: dim.Precision, // Only divide by precision to restore decimals |
| 266 | } |
| 267 | chart.AddDim(chartDim) |
| 268 | } |
| 269 | |
| 270 | // Apply global labels to the new chart |
| 271 | c.applyGlobalLabelsToChart(chart) |
| 272 | |
| 273 | return chart |
| 274 | } |
| 275 | |
| 276 | // cleanChartID converts context name to valid chart ID |
| 277 | func cleanChartID(contextName string) string { |
| 278 | // Convert dots to underscores for chart ID |
| 279 | // example.test_absolute -> example_test_absolute |
| 280 | return strings.ReplaceAll(contextName, ".", "_") |
| 281 | } |
| 282 | |
| 283 | func (c *Collector) markInstanceChartsPresent(instanceKey string) { |
| 284 | if c.instanceCharts == nil { |
| 285 | c.instanceCharts = make(map[string]struct{}) |
| 286 | } |
| 287 | c.instanceCharts[instanceKey] = struct{}{} |
| 288 | } |
| 289 | |
| 290 | // hasChartsForInstance checks if charts exist for a given instance |
| 291 | func (c *Collector) hasChartsForInstance(instanceKey string) bool { |
| 292 | _, ok := c.instanceCharts[instanceKey] |
| 293 | return ok |
| 294 | } |
| 295 | |
| 296 | // markChartsObsolete marks all charts for a given instance as obsolete |
| 297 | func (c *Collector) markChartsObsolete(instanceKey string) { |
| 298 | // Find charts that belong to this instance |
| 299 | chartID := cleanChartID(instanceKey) |
| 300 | |
| 301 | for _, chart := range *c.charts { |
| 302 | if chart.ID == chartID && !chart.Obsolete { |
| 303 | c.Debugf("Marking chart %s as obsolete for instance %s", chart.ID, instanceKey) |
| 304 | chart.Obsolete = true |
| 305 | chart.MarkNotCreated() // Reset created flag to trigger CHART command with obsolete flag |
| 306 | } |
| 307 | } |
| 308 | } |
| 309 | |
| 310 | // Helper method for collectors to register generated contexts |
| 311 | func (c *Collector) RegisterContexts(contexts ...any) { |
| 312 | c.registeredContexts = append(c.registeredContexts, contexts...) |
| 313 | // Build context map for quick lookup |
| 314 | if c.contextMap == nil { |
| 315 | c.contextMap = make(map[string]any) |
| 316 | } |
| 317 | for _, ctx := range contexts { |
| 318 | name := extractContextName(ctx) |
| 319 | c.contextMap[name] = ctx |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | // GetBase returns the base module |
| 324 | func (c *Collector) GetBase() *collectorapi.Base { |
| 325 | return &c.Base |
| 326 | } |
| 327 | |
| 328 | // Configuration returns the configuration interface |
| 329 | func (c *Collector) Configuration() any { |
| 330 | // Default implementation - collectors should override this |
| 331 | return c.Config |
| 332 | } |
| 333 | |
| 334 | // SetGlobalLabel adds or updates a job-level label that will be applied to all charts |
| 335 | func (c *Collector) SetGlobalLabel(key, value string) { |
| 336 | // Check if label already exists |
| 337 | for i, label := range c.globalLabels { |
| 338 | if label.Key == key { |
| 339 | c.globalLabels[i].Value = value |
| 340 | c.updateChartsWithGlobalLabels() |
| 341 | return |
| 342 | } |
| 343 | } |
| 344 | |
| 345 | // Add new label |
| 346 | c.globalLabels = append(c.globalLabels, collectorapi.Label{Key: key, Value: value}) |
| 347 | c.updateChartsWithGlobalLabels() |
| 348 | } |
| 349 | |
| 350 | // SetGlobalLabels replaces all global labels |
| 351 | func (c *Collector) SetGlobalLabels(labels map[string]string) { |
| 352 | c.globalLabels = make([]collectorapi.Label, 0, len(labels)) |
| 353 | for key, value := range labels { |
| 354 | c.globalLabels = append(c.globalLabels, collectorapi.Label{Key: key, Value: value}) |
| 355 | } |
| 356 | c.updateChartsWithGlobalLabels() |
| 357 | } |
| 358 | |
| 359 | // updateChartsWithGlobalLabels applies global labels to all existing charts |
| 360 | func (c *Collector) updateChartsWithGlobalLabels() { |
| 361 | if c.charts == nil { |
| 362 | return |
| 363 | } |
| 364 | |
| 365 | for _, chart := range *c.charts { |
| 366 | // Update existing chart labels |
| 367 | c.applyGlobalLabelsToChart(chart) |
| 368 | } |
| 369 | } |
| 370 | |
| 371 | // applyGlobalLabelsToChart adds global labels to a chart, avoiding duplicates |
| 372 | func (c *Collector) applyGlobalLabelsToChart(chart *collectorapi.Chart) { |
| 373 | // Create a map of existing labels for quick lookup |
| 374 | existingLabels := make(map[string]int) |
| 375 | for i, label := range chart.Labels { |
| 376 | existingLabels[label.Key] = i |
| 377 | } |
| 378 | |
| 379 | // Apply global labels |
| 380 | for _, globalLabel := range c.globalLabels { |
| 381 | if idx, exists := existingLabels[globalLabel.Key]; exists { |
| 382 | // Update existing label |
| 383 | chart.Labels[idx] = globalLabel |
| 384 | } else { |
| 385 | // Add new label |
| 386 | chart.Labels = append(chart.Labels, globalLabel) |
| 387 | } |
| 388 | } |
| 389 | } |
| 390 | |
| 391 | // GetCurrentIteration returns the current global iteration counter |
| 392 | func (c *Collector) GetCurrentIteration() int64 { |
| 393 | if c.State != nil { |
| 394 | return c.State.GetIteration() |
| 395 | } |
| 396 | return 0 |
| 397 | } |