Skip to main content

pipeline_builder.go

pipeline_builder.go - Overview​

  1. Overview This file defines the logic for building and updating processing pipelines for metrics and traces in an OpAMP agent. It manages the order and enablement of processors within these pipelines, ensuring no duplicates and handling conflicts between existing configurations and desired specifications.

  2. Detailed Documentation

Data Structures​

pipelineStatus​

type pipelineStatus struct {
	Name    string
	Enabled bool
}
  • Purpose: Represents the status of a processor within a pipeline.
    • Parameters:
      • Name (string): The name of the processor.
      • Enabled (bool): Indicates whether the processor is enabled.
    • Returns: None

Variables​

lockTracesPipelineSpec​

var lockTracesPipelineSpec sync.RWMutex
  • Purpose: A read-write mutex to protect concurrent access to the tracesPipelineSpec variable.

lockMetricsPipelineSpec​

var lockMetricsPipelineSpec sync.RWMutex
  • Purpose: A read-write mutex to protect concurrent access to the metricsPipelineSpec variable.

tracesPipelineSpec​

var tracesPipelineSpec = map[int]pipelineStatus{
	0: {
		Name:    "signoz_tail_sampling",
		Enabled: false,
	},
	1: {
		Name:    "batch",
		Enabled: true,
	},
}
  • Purpose: Defines the default processing pipeline specification for traces. It is a map of integer indices to pipelineStatus structs.
    • Parameters: None
    • Returns: None

metricsPipelineSpec​

var metricsPipelineSpec = map[int]pipelineStatus{
	0: {
		Name:    "filter",
		Enabled: false,
	},
	1: {
		Name:    "batch",
		Enabled: true,
	},
}
  • Purpose: Defines the default processing pipeline specification for metrics. It is a map of integer indices to pipelineStatus structs.
    • Parameters: None
    • Returns: None

Functions​

updatePipelineSpec​

func updatePipelineSpec(signal string, name string, enabled bool)
  • Purpose: Updates the enabled status of a processor in either the metrics or traces pipeline specification.
    • Parameters:
      • signal (string): The signal type ("metrics" or "traces").
      • name (string): The name of the processor to update.
      • enabled (bool): The new enabled status for the processor.
    • Returns: None

AddToTracePipelineSpec​

func AddToTracePipelineSpec(processor string)
  • Purpose: Enables a processor in the traces pipeline specification.
    • Parameters:
      • processor (string): The name of the processor to enable.
    • Returns: None

RemoveFromTracePipelineSpec​

func RemoveFromTracePipelineSpec(name string)
  • Purpose: Disables a processor in the traces pipeline specification.
    • Parameters:
      • name (string): The name of the processor to disable.
    • Returns: None

AddToMetricsPipelineSpec​

func AddToMetricsPipelineSpec(processor string)
  • Purpose: Enables a processor in the metrics pipeline specification.
    • Parameters:
      • processor (string): The name of the processor to enable.
    • Returns: None

RemoveFromMetricsPipelineSpec​

func RemoveFromMetricsPipelineSpec(name string)
  • Purpose: Disables a processor in the metrics pipeline specification.
    • Parameters:
      • name (string): The name of the processor to disable.
    • Returns: None

checkDuplicates​

func checkDuplicates(pipeline []interface{}) bool
  • Purpose: Checks if there are duplicate processor names within a pipeline.
    • Parameters:
      • pipeline ([]interface{}), A slice of interface{} representing the pipeline processors.
    • Returns:
      • (bool): true if duplicates are found, false otherwise.

buildPipeline​

func buildPipeline(signal Signal, current []interface{}) ([]interface{}, error)
  • Purpose: Builds a processing pipeline based on the given signal type (metrics or traces) and the current existing pipeline configuration. It reconciles the desired pipeline specification with the existing configuration, adding, removing, or reordering processors as necessary.
    • Parameters:
      • signal (Signal): The signal type (Metrics or Traces). Signal type not defined in this file.
      • current ([]interface{}): The current pipeline configuration as a slice of interface{}. Each element is assumed to be a processor name (string).
    • Returns:
      • ([]interface{}): The updated pipeline configuration as a slice of interface{}.
      • (error): An error if the signal is invalid or if the resulting pipeline contains duplicates due to configuration conflicts.

  1. Code Examples N/A

  2. Clarity and Accuracy The documentation is based on the code provided and avoids making assumptions or adding extra context.

  3. Markdown & MDX Perfection The markdown syntax is correct, and special characters are escaped.

  4. Edge Cases To Avoid Breaking MDX All potential MDX breaking characters are escaped or wrapped in backticks.

  5. Getting Started Relevance Include in Getting Started: NO