feat(loki/podlogs): add pipelineStages processing to PodLogs CRD

internal/component/loki/process/metric/counters.go

@@ -18,17 +18,17 @@ const (
 // CounterConfig defines a counter metric whose value only goes up.
 type CounterConfig struct {
 	// Shared fields
-	Name        string        `alloy:"name,attr"`
-	Description string        `alloy:"description,attr,optional"`
-	Source      string        `alloy:"source,attr,optional"`
-	Prefix      string        `alloy:"prefix,attr,optional"`
-	MaxIdle     time.Duration `alloy:"max_idle_duration,attr,optional"`
-	Value       string        `alloy:"value,attr,optional"`
+	Name        string        `alloy:"name,attr"                       json:"name"`
+	Description string        `alloy:"description,attr,optional"       json:"description,omitempty"`
+	Source      string        `alloy:"source,attr,optional"            json:"source,omitempty"`
+	Prefix      string        `alloy:"prefix,attr,optional"            json:"prefix,omitempty"`
+	MaxIdle     time.Duration `alloy:"max_idle_duration,attr,optional" json:"-"` // not supported in CRD: time.Duration is nanosecond int64 in JSON
+	Value       string        `alloy:"value,attr,optional"             json:"value,omitempty"`
 
 	// Counter-specific fields
-	Action          string `alloy:"action,attr"`
-	MatchAll        bool   `alloy:"match_all,attr,optional"`
-	CountEntryBytes bool   `alloy:"count_entry_bytes,attr,optional"`
+	Action          string `alloy:"action,attr"                    json:"action"`
+	MatchAll        bool   `alloy:"match_all,attr,optional"        json:"matchAll,omitempty"`
+	CountEntryBytes bool   `alloy:"count_entry_bytes,attr,optional" json:"countEntryBytes,omitempty"`
 }
 
 // DefaultCounterConfig sets the default for a Counter.

internal/component/loki/process/metric/gauges.go

@@ -29,15 +29,15 @@ var DefaultGaugeConfig = GaugeConfig{
 // GaugeConfig defines a gauge metric whose value can go up or down.
 type GaugeConfig struct {
 	// Shared fields
-	Name        string        `alloy:"name,attr"`
-	Description string        `alloy:"description,attr,optional"`
-	Source      string        `alloy:"source,attr,optional"`
-	Prefix      string        `alloy:"prefix,attr,optional"`
-	MaxIdle     time.Duration `alloy:"max_idle_duration,attr,optional"`
-	Value       string        `alloy:"value,attr,optional"`
+	Name        string        `alloy:"name,attr"                       json:"name"`
+	Description string        `alloy:"description,attr,optional"       json:"description,omitempty"`
+	Source      string        `alloy:"source,attr,optional"            json:"source,omitempty"`
+	Prefix      string        `alloy:"prefix,attr,optional"            json:"prefix,omitempty"`
+	MaxIdle     time.Duration `alloy:"max_idle_duration,attr,optional" json:"-"` // not supported in CRD: time.Duration is nanosecond int64 in JSON
+	Value       string        `alloy:"value,attr,optional"             json:"value,omitempty"`
 
 	// Gauge-specific fields
-	Action string `alloy:"action,attr"`
+	Action string `alloy:"action,attr" json:"action"`
 }
 
 // SetToDefault implements syntax.Defaulter.

internal/component/loki/process/metric/histograms.go

@@ -18,15 +18,15 @@ var DefaultHistogramConfig = HistogramConfig{
 // HistogramConfig defines a histogram metric whose values are bucketed.
 type HistogramConfig struct {
 	// Shared fields
-	Name        string        `alloy:"name,attr"`
-	Description string        `alloy:"description,attr,optional"`
-	Source      string        `alloy:"source,attr,optional"`
-	Prefix      string        `alloy:"prefix,attr,optional"`
-	MaxIdle     time.Duration `alloy:"max_idle_duration,attr,optional"`
-	Value       string        `alloy:"value,attr,optional"`
+	Name        string        `alloy:"name,attr"                       json:"name"`
+	Description string        `alloy:"description,attr,optional"       json:"description,omitempty"`
+	Source      string        `alloy:"source,attr,optional"            json:"source,omitempty"`
+	Prefix      string        `alloy:"prefix,attr,optional"            json:"prefix,omitempty"`
+	MaxIdle     time.Duration `alloy:"max_idle_duration,attr,optional" json:"-"` // not supported in CRD: time.Duration is nanosecond int64 in JSON
+	Value       string        `alloy:"value,attr,optional"             json:"value,omitempty"`
 
 	// Histogram-specific fields
-	Buckets []float64 `alloy:"buckets,attr"`
+	Buckets []float64 `alloy:"buckets,attr" json:"buckets"`
 }
 
 // SetToDefault implements syntax.Defaulter.

internal/component/loki/process/stages/cri.go

@@ -19,9 +19,9 @@ import (
 // CRIConfig is an empty struct that is used to enable a pre-defined pipeline
 // for decoding entries that are using the CRI logging format.
 type CRIConfig struct {
-	MaxPartialLines            int    `alloy:"max_partial_lines,attr,optional"`
-	MaxPartialLineSize         uint64 `alloy:"max_partial_line_size,attr,optional"`
-	MaxPartialLineSizeTruncate bool   `alloy:"max_partial_line_size_truncate,attr,optional"`
+	MaxPartialLines            int    `alloy:"max_partial_lines,attr,optional"             json:"maxPartialLines,omitempty"`
+	MaxPartialLineSize         uint64 `alloy:"max_partial_line_size,attr,optional"         json:"maxPartialLineSize,omitempty"`
+	MaxPartialLineSizeTruncate bool   `alloy:"max_partial_line_size_truncate,attr,optional" json:"maxPartialLineSizeTruncate,omitempty"`
 }
 
 var (

internal/component/loki/process/stages/drop.go

@@ -31,13 +31,13 @@ var (
 
 // DropConfig contains the configuration for a dropStage
 type DropConfig struct {
-	DropReason string           `alloy:"drop_counter_reason,attr,optional"`
-	Source     string           `alloy:"source,attr,optional"`
-	Value      string           `alloy:"value,attr,optional"`
-	Separator  string           `alloy:"separator,attr,optional"`
-	Expression string           `alloy:"expression,attr,optional"`
-	OlderThan  time.Duration    `alloy:"older_than,attr,optional"`
-	LongerThan units.Base2Bytes `alloy:"longer_than,attr,optional"`
+	DropReason string           `alloy:"drop_counter_reason,attr,optional" json:"dropReason,omitempty"`
+	Source     string           `alloy:"source,attr,optional"              json:"source,omitempty"`
+	Value      string           `alloy:"value,attr,optional"               json:"value,omitempty"`
+	Separator  string           `alloy:"separator,attr,optional"           json:"separator,omitempty"`
+	Expression string           `alloy:"expression,attr,optional"          json:"expression,omitempty"`
+	OlderThan  time.Duration    `alloy:"older_than,attr,optional"          json:"-"`                    // not supported in CRD: time.Duration is nanosecond int64 in JSON
+	LongerThan units.Base2Bytes `alloy:"longer_than,attr,optional"         json:"longerThan,omitempty"` // TextMarshaler: serializes as "5MiB"
 }
 
 // validateDropConfig validates the DropConfig for the dropStage

internal/component/loki/process/stages/geoip.go

@@ -56,10 +56,10 @@ var fields = map[GeoIPFields]string{
 
 // GeoIPConfig represents GeoIP stage config
 type GeoIPConfig struct {
-	DB            string            `alloy:"db,attr"`
-	Source        *string           `alloy:"source,attr"`
-	DBType        string            `alloy:"db_type,attr,optional"`
-	CustomLookups map[string]string `alloy:"custom_lookups,attr,optional"`
+	DB            string            `alloy:"db,attr"                          json:"db"`
+	Source        *string           `alloy:"source,attr"                      json:"source"`
+	DBType        string            `alloy:"db_type,attr,optional"            json:"dbType,omitempty"`
+	CustomLookups map[string]string `alloy:"custom_lookups,attr,optional"     json:"customLookups,omitempty"`
 }
 
 func validateGeoIPConfig(c GeoIPConfig) (map[string]jmespath.JMESPath, error) {

internal/component/loki/process/stages/json.go

@@ -22,9 +22,9 @@ const (
 
 // JSONConfig represents a JSON Stage configuration
 type JSONConfig struct {
-	Expressions   map[string]string `alloy:"expressions,attr"`
-	Source        *string           `alloy:"source,attr,optional"`
-	DropMalformed bool              `alloy:"drop_malformed,attr,optional"`
+	Expressions   map[string]string `alloy:"expressions,attr"              json:"expressions"`
+	Source        *string           `alloy:"source,attr,optional"          json:"source,omitempty"`
+	DropMalformed bool              `alloy:"drop_malformed,attr,optional"  json:"dropMalformed,omitempty"`
 }
 
 // validateJSONConfig validates a json config and returns a map of necessary jmespath expressions.

internal/component/loki/process/stages/label_drop.go

@@ -12,7 +12,7 @@ var ErrEmptyLabelDropStageConfig = errors.New("labeldrop stage config cannot be
 
 // LabelDropConfig contains the slice of labels to be dropped.
 type LabelDropConfig struct {
-	Values []string `alloy:"values,attr"`
+	Values []string `alloy:"values,attr" json:"values"`
 }
 
 func newLabelDropStage(config LabelDropConfig) (Stage, error) {

internal/component/loki/process/stages/label_keep.go

@@ -12,7 +12,7 @@ var ErrEmptyLabelAllowStageConfig = errors.New("labelallow stage config cannot b
 
 // LabelAllowConfig contains the slice of labels to allow through.
 type LabelAllowConfig struct {
-	Values []string `alloy:"values,attr"`
+	Values []string `alloy:"values,attr" json:"values"`
 }
 
 func newLabelAllowStage(config LabelAllowConfig) (Stage, error) {

internal/component/loki/process/stages/labels.go

@@ -23,8 +23,8 @@ const (
 
 // LabelsConfig is a set of labels to be extracted
 type LabelsConfig struct {
-	Values     map[string]*string `alloy:"values,attr"`
-	SourceType SourceType         `alloy:"source_type,attr,optional"`
+	Values     map[string]*string `alloy:"values,attr"                json:"values"`
+	SourceType SourceType         `alloy:"source_type,attr,optional"  json:"sourceType,omitempty"`
 }
 
 // validateLabelsConfig validates the Label stage configuration

internal/component/loki/process/stages/limit.go

@@ -24,11 +24,11 @@ const MinReasonableMaxDistinctLabels = 10000 // 80bytes per rate.Limiter ~ 1MiB
 
 // LimitConfig sets up a Limit stage.
 type LimitConfig struct {
-	Rate              float64 `alloy:"rate,attr"`
-	Burst             int     `alloy:"burst,attr"`
-	Drop              bool    `alloy:"drop,attr,optional"`
-	ByLabelName       string  `alloy:"by_label_name,attr,optional"`
-	MaxDistinctLabels int     `alloy:"max_distinct_labels,attr,optional"`
+	Rate              float64 `alloy:"rate,attr"                        json:"rate"`
+	Burst             int     `alloy:"burst,attr"                       json:"burst"`
+	Drop              bool    `alloy:"drop,attr,optional"               json:"drop,omitempty"`
+	ByLabelName       string  `alloy:"by_label_name,attr,optional"      json:"byLabelName,omitempty"`
+	MaxDistinctLabels int     `alloy:"max_distinct_labels,attr,optional" json:"maxDistinctLabels,omitempty"`
 }
 
 func newLimitStage(logger log.Logger, cfg LimitConfig, registerer prometheus.Registerer) (Stage, error) {

internal/component/loki/process/stages/logfmt.go

@@ -20,8 +20,8 @@ var (
 
 // LogfmtConfig represents a logfmt Stage configuration
 type LogfmtConfig struct {
-	Mapping map[string]string `alloy:"mapping,attr"`
-	Source  string            `alloy:"source,attr,optional"`
+	Mapping map[string]string `alloy:"mapping,attr"          json:"mapping"`
+	Source  string            `alloy:"source,attr,optional"  json:"source,omitempty"`
 }
 
 // validateLogfmtConfig validates a logfmt stage config and returns an inverse mapping of configured mapping.

internal/component/loki/process/stages/luhn.go

@@ -11,10 +11,10 @@ import (
 
 // LuhnFilterConfig configures a processing stage that filters out Luhn-valid numbers.
 type LuhnFilterConfig struct {
-	Replacement string  `alloy:"replacement,attr,optional"`
-	Source      *string `alloy:"source,attr,optional"`
-	MinLength   int     `alloy:"min_length,attr,optional"`
-	Delimiters  string  `alloy:"delimiters,attr,optional"`
+	Replacement string  `alloy:"replacement,attr,optional"  json:"replacement,omitempty"`
+	Source      *string `alloy:"source,attr,optional"       json:"source,omitempty"`
+	MinLength   int     `alloy:"min_length,attr,optional"   json:"minLength,omitempty"`
+	Delimiters  string  `alloy:"delimiters,attr,optional"   json:"delimiters,omitempty"`
 }
 
 // validateLuhnFilterConfig validates the LuhnFilterConfig.

internal/component/loki/process/stages/match.go

@@ -27,11 +27,11 @@ var (
 
 // MatchConfig contains the configuration for a matcherStage
 type MatchConfig struct {
-	Selector     string        `alloy:"selector,attr"`
-	Stages       []StageConfig `alloy:"stage,enum,optional"`
-	Action       string        `alloy:"action,attr,optional"`
-	PipelineName string        `alloy:"pipeline_name,attr,optional"`
-	DropReason   string        `alloy:"drop_counter_reason,attr,optional"`
+	Selector     string        `alloy:"selector,attr"                     json:"selector"`
+	Stages       []StageConfig `alloy:"stage,enum,optional"               json:"stages,omitempty"`
+	Action       string        `alloy:"action,attr,optional"              json:"action,omitempty"`
+	PipelineName string        `alloy:"pipeline_name,attr,optional"       json:"pipelineName,omitempty"`
+	DropReason   string        `alloy:"drop_counter_reason,attr,optional" json:"dropReason,omitempty"`
 }
 
 // validateMatcherConfig validates the MatcherConfig for the matcherStage

internal/component/loki/process/stages/metric.go

@@ -22,14 +22,14 @@ const (
 
 // MetricConfig is a single metrics configuration.
 type MetricConfig struct {
-	Counter   *metric.CounterConfig   `alloy:"counter,block,optional"`
-	Gauge     *metric.GaugeConfig     `alloy:"gauge,block,optional"`
-	Histogram *metric.HistogramConfig `alloy:"histogram,block,optional"`
+	Counter   *metric.CounterConfig   `alloy:"counter,block,optional"   json:"counter,omitempty"`
+	Gauge     *metric.GaugeConfig     `alloy:"gauge,block,optional"     json:"gauge,omitempty"`
+	Histogram *metric.HistogramConfig `alloy:"histogram,block,optional" json:"histogram,omitempty"`
 }
 
 // MetricsConfig is a set of configured metrics.
 type MetricsConfig struct {
-	Metrics []MetricConfig `alloy:"metric,enum,optional"`
+	Metrics []MetricConfig `alloy:"metric,enum,optional" json:"metrics,omitempty"`
 }
 
 type cfgCollector struct {

internal/component/loki/process/stages/output.go

@@ -19,7 +19,7 @@ var (
 // OutputConfig initializes a configuration stage which sets the log line to a
 // value from the extracted map.
 type OutputConfig struct {
-	Source string `alloy:"source,attr"`
+	Source string `alloy:"source,attr" json:"source"`
 }
 
 // newOutputStage creates a new outputStage

internal/component/loki/process/stages/pack.go

@@ -105,8 +105,8 @@ func (w Packed) MarshalJSON() ([]byte, error) {
 
 // PackConfig contains the configuration for a packStage
 type PackConfig struct {
-	Labels          []string `alloy:"labels,attr"`
-	IngestTimestamp bool     `alloy:"ingest_timestamp,attr,optional"`
+	Labels          []string `alloy:"labels,attr"                    json:"labels"`
+	IngestTimestamp bool     `alloy:"ingest_timestamp,attr,optional" json:"ingestTimestamp,omitempty"`
 }
 
 // DefaultPackConfig sets the defaults.

internal/component/loki/process/stages/pattern.go

@@ -23,9 +23,9 @@ var (
 // extract values from log lines into the shared values map.
 // See https://grafana.com/docs/loki/latest/query/log_queries/#pattern
 type PatternConfig struct {
-	Pattern          string  `alloy:"pattern,attr"`
-	Source           *string `alloy:"source,attr,optional"`
-	LabelsFromGroups bool    `alloy:"labels_from_groups,attr,optional"`
+	Pattern          string  `alloy:"pattern,attr"                   json:"pattern"`
+	Source           *string `alloy:"source,attr,optional"           json:"source,omitempty"`
+	LabelsFromGroups bool    `alloy:"labels_from_groups,attr,optional" json:"labelsFromGroups,omitempty"`
 }
 
 // validatePatternConfig validates the config and return a regex