Merge branch 'main' into subgraph-for-module-components

CHANGELOG.md

@@ -63,6 +63,8 @@ Main (unreleased)
 - Added additional http client proxy configurations to components for
   `no_proxy`, `proxy_from_environment`, and `proxy_connect_header`. (@erikbaranowski)
 
+- Batch staleness tracking to reduce mutex contention and increase performance. (@mattdurham)
+
 ### Bugfixes
 
 - Fix an issue in `remote.s3` where the exported content of an object would be an empty string if `remote.s3` failed to fully retrieve
@@ -97,6 +99,9 @@ Main (unreleased)
 
 - Updated docs for MSSQL Integration to show additional authentication capabilities. (@StefanKurek)
 
+- `grafana-agent` and `grafana-agent-flow` fallback to default X.509 trusted root certificates
+  when the `GODEBUG=x509usefallbackroots=1` environment variable is set. (@hainenber)
+
 v0.39.2 (2024-1-31)
 --------------------
 

cmd/grafana-agent-flow/main.go

@@ -17,6 +17,10 @@ import (
 
 	// Register integrations
 	_ "github.com/grafana/agent/pkg/integrations/install"
+
+	// Embed a set of fallback X.509 trusted roots
+	// Allows the app to work correctly even when the OS does not provide a verifier or systems roots pool
+	_ "golang.org/x/crypto/x509roots/fallback"
 )
 
 func init() {

cmd/grafana-agent/main.go

@@ -21,6 +21,10 @@ import (
 
 	// Register integrations
 	_ "github.com/grafana/agent/pkg/integrations/install"
+
+	// Embed a set of fallback X.509 trusted roots
+	// Allows the app to work correctly even when the OS does not provide a verifier or systems roots pool
+	_ "golang.org/x/crypto/x509roots/fallback"
 )
 
 func init() {

component/prometheus/fanout.go

@@ -12,7 +12,6 @@ import (
 	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/metadata"
-	"github.com/prometheus/prometheus/model/value"
 	"github.com/prometheus/prometheus/scrape"
 	"github.com/prometheus/prometheus/storage"
 )
@@ -75,11 +74,12 @@ func (f *Fanout) Appender(ctx context.Context) storage.Appender {
 	ctx = scrape.ContextWithMetricMetadataStore(ctx, NoopMetadataStore{})
 
 	app := &appender{
-		children:       make([]storage.Appender, 0),
-		componentID:    f.componentID,
-		writeLatency:   f.writeLatency,
-		samplesCounter: f.samplesCounter,
-		ls:             f.ls,
+		children:          make([]storage.Appender, 0),
+		componentID:       f.componentID,
+		writeLatency:      f.writeLatency,
+		samplesCounter:    f.samplesCounter,
+		ls:                f.ls,
+		stalenessTrackers: make([]labelstore.StalenessTracker, 0),
 	}
 
 	for _, x := range f.children {
@@ -92,12 +92,13 @@ func (f *Fanout) Appender(ctx context.Context) storage.Appender {
 }
 
 type appender struct {
-	children       []storage.Appender
-	componentID    string
-	writeLatency   prometheus.Histogram
-	samplesCounter prometheus.Counter
-	start          time.Time
-	ls             labelstore.LabelStore
+	children          []storage.Appender
+	componentID       string
+	writeLatency      prometheus.Histogram
+	samplesCounter    prometheus.Counter
+	start             time.Time
+	ls                labelstore.LabelStore
+	stalenessTrackers []labelstore.StalenessTracker
 }
 
 var _ storage.Appender = (*appender)(nil)
@@ -110,12 +111,11 @@ func (a *appender) Append(ref storage.SeriesRef, l labels.Labels, t int64, v flo
 	if ref == 0 {
 		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
 	}
-	if value.IsStaleNaN(v) {
-		a.ls.AddStaleMarker(uint64(ref), l)
-	} else {
-		// Tested this to ensure it had no cpu impact, since it is called so often.
-		a.ls.RemoveStaleMarker(uint64(ref))
-	}
+	a.stalenessTrackers = append(a.stalenessTrackers, labelstore.StalenessTracker{
+		GlobalRefID: uint64(ref),
+		Labels:      l,
+		Value:       v,
+	})
 	var multiErr error
 	updated := false
 	for _, x := range a.children {
@@ -136,6 +136,7 @@ func (a *appender) Append(ref storage.SeriesRef, l labels.Labels, t int64, v flo
 func (a *appender) Commit() error {
 	defer a.recordLatency()
 	var multiErr error
+	a.ls.TrackStaleness(a.stalenessTrackers)
 	for _, x := range a.children {
 		err := x.Commit()
 		if err != nil {
@@ -148,6 +149,7 @@ func (a *appender) Commit() error {
 // Rollback satisfies the Appender interface.
 func (a *appender) Rollback() error {
 	defer a.recordLatency()
+	a.ls.TrackStaleness(a.stalenessTrackers)
 	var multiErr error
 	for _, x := range a.children {
 		err := x.Rollback()

component/prometheus/interceptor.go

@@ -8,7 +8,6 @@ import (
 	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/metadata"
-	"github.com/prometheus/prometheus/model/value"
 	"github.com/prometheus/prometheus/storage"
 )
 
@@ -80,8 +79,9 @@ func WithHistogramHook(f func(ref storage.SeriesRef, l labels.Labels, t int64, h
 // Appender satisfies the Appendable interface.
 func (f *Interceptor) Appender(ctx context.Context) storage.Appender {
 	app := &interceptappender{
-		interceptor: f,
-		ls:          f.ls,
+		interceptor:       f,
+		ls:                f.ls,
+		stalenessTrackers: make([]labelstore.StalenessTracker, 0),
 	}
 	if f.next != nil {
 		app.child = f.next.Appender(ctx)
@@ -90,9 +90,10 @@ func (f *Interceptor) Appender(ctx context.Context) storage.Appender {
 }
 
 type interceptappender struct {
-	interceptor *Interceptor
-	child       storage.Appender
-	ls          labelstore.LabelStore
+	interceptor       *Interceptor
+	child             storage.Appender
+	ls                labelstore.LabelStore
+	stalenessTrackers []labelstore.StalenessTracker
 }
 
 var _ storage.Appender = (*interceptappender)(nil)
@@ -102,13 +103,11 @@ func (a *interceptappender) Append(ref storage.SeriesRef, l labels.Labels, t int
 	if ref == 0 {
 		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
 	}
-
-	if value.IsStaleNaN(v) {
-		a.ls.AddStaleMarker(uint64(ref), l)
-	} else {
-		// Tested this to ensure it had no cpu impact, since it is called so often.
-		a.ls.RemoveStaleMarker(uint64(ref))
-	}
+	a.stalenessTrackers = append(a.stalenessTrackers, labelstore.StalenessTracker{
+		GlobalRefID: uint64(ref),
+		Labels:      l,
+		Value:       v,
+	})
 
 	if a.interceptor.onAppend != nil {
 		return a.interceptor.onAppend(ref, l, t, v, a.child)
@@ -121,6 +120,7 @@ func (a *interceptappender) Append(ref storage.SeriesRef, l labels.Labels, t int
 
 // Commit satisfies the Appender interface.
 func (a *interceptappender) Commit() error {
+	a.ls.TrackStaleness(a.stalenessTrackers)
 	if a.child == nil {
 		return nil
 	}
@@ -129,6 +129,7 @@ func (a *interceptappender) Commit() error {
 
 // Rollback satisfies the Appender interface.
 func (a *interceptappender) Rollback() error {
+	a.ls.TrackStaleness(a.stalenessTrackers)
 	if a.child == nil {
 		return nil
 	}

docs/sources/flow/reference/components/loki.process.md

@@ -961,7 +961,7 @@ The following arguments are supported:
 | Name               | Type           | Description                                                                     | Default | Required |
 | ------------------ | -------------- | ------------------------------------------------------------------------------- | ------- | -------- |
 | `labels`           | `list(string)` | The values from the extracted data and labels to pack with the log entry.       |         | yes      |
-| `ingest_timestamp` | `bool`         | Whether to replace the log entry timestamp with the time the `pack` stage runs. | `true   | no       |
+| `ingest_timestamp` | `bool`         | Whether to replace the log entry timestamp with the time the `pack` stage runs. | `true`  | no       |
 
 This stage lets you embed extracted values and labels together with the log
 line, by packing them into a JSON object. The original message is stored under
@@ -1459,6 +1459,11 @@ The following arguments are supported:
 | `location`          | `string`       | IANA Timezone Database location to use when parsing.        | `""`      | no       |
 | `action_on_failure` | `string`       | What to do when the timestamp can't be extracted or parsed. | `"fudge"` | no       |
 
+{{< admonition type="note" >}}
+Be careful with further stages which may also override the timestamp.
+For example, a `stage.pack` with `ingest_timestamp` set to `true` could replace the timestamp which `stage.timestamp` had set earlier in the pipeline.
+{{< /admonition >}}
+
 The `source` field defines which value from the shared map of extracted values
 the stage should attempt to parse as a timestamp.
 

docs/sources/flow/reference/components/otelcol.auth.headers.md

@@ -70,9 +70,12 @@ Exactly one of `value` or `from_context` must be provided for each `header`
 block.
 
 The `value` attribute sets the value of the header directly.
+Alternatively, `from_context` can be used to dynamically retrieve the header value from request metadata.
 
-Alternatively, `from_context` can be used to dynamically retrieve the header
-value from request metadata.
+For `from_context` to work, other components in the pipeline also need to be configured appropriately:
+* If an `otelcol.processor.batch` is present in the pipeline, it must be configured to preserve client metadata. 
+  Do this by adding the value that `from_context` needs to the `metadata_keys` of the batch processor.
+* `otelcol` receivers must be configured with `include_metadata` set to `true` so that metadata keys are available to the pipeline.
 
 ## Exported fields
 
@@ -96,10 +99,29 @@ configuration.
 This example configures [otelcol.exporter.otlp][] to use custom headers:
 
 ```river
-otelcol.exporter.otlp "example" {
-  client {
-    endpoint = "my-otlp-grpc-server:4317"
-    auth     = otelcol.auth.headers.creds.handler
+otelcol.receiver.otlp "default" {
+  http {
+    include_metadata = true
+  }
+  grpc {
+    include_metadata = true
+  }
+
+  output {
+    metrics = [otelcol.processor.batch.default.input]
+    logs    = [otelcol.processor.batch.default.input]
+    traces  = [otelcol.processor.batch.default.input]
+  }
+}
+
+otelcol.processor.batch "default" {
+  // Preserve the tenant_id metadata.
+  metadata_keys = ["tenant_id"]
+
+  output {
+    metrics = [otelcol.exporter.otlp.production.input]
+    logs    = [otelcol.exporter.otlp.production.input]
+    traces  = [otelcol.exporter.otlp.production.input]
   }
 }
 
@@ -114,6 +136,13 @@ otelcol.auth.headers "creds" {
     value = "user_id"
   }
 }
+
+otelcol.exporter.otlp "production" {
+  client {
+    endpoint = env("OTLP_SERVER_ENDPOINT")
+    auth     = otelcol.auth.headers.creds.handler
+  }
+}
 ```
 
 [otelcol.exporter.otlp]: {{< relref "./otelcol.exporter.otlp.md" >}}

go.mod

@@ -616,6 +616,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.87.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/vcenterreceiver v0.87.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.42.0
+	golang.org/x/crypto/x509roots/fallback v0.0.0-20240208163226-62c9f1799c91
 	k8s.io/apimachinery v0.28.3
 )
 

go.sum

@@ -2513,6 +2513,8 @@ golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU
 golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I=
 golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
 golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
+golang.org/x/crypto/x509roots/fallback v0.0.0-20240208163226-62c9f1799c91 h1:Lyizcy9jX02jYR0ceBkL6S+jRys8Uepf7wt1vrz6Ras=
+golang.org/x/crypto/x509roots/fallback v0.0.0-20240208163226-62c9f1799c91/go.mod h1:kNa9WdvYnzFwC79zRpLRMJbdEFlhyM5RPFBBZp/wWH8=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=

operations/helm/charts/grafana-agent/CHANGELOG.md

@@ -10,6 +10,15 @@ internal API changes are not present.
 Unreleased
 ----------
 
+0.32.0 (2024-02-15)
+-------------------
+
+### Enhancements
+
+- Allow setting scheme for readiness checks when using tls. (@captncraig)
+
+- Update Grafana Agent version to v0.39.2. (@captncraig)
+
 0.31.1 (2024-01-19)
 -------------------
 

operations/helm/charts/grafana-agent/Chart.yaml

@@ -2,8 +2,8 @@ apiVersion: v2
 name: grafana-agent
 description: 'Grafana Agent'
 type: application
-version: 0.31.1
-appVersion: 'v0.39.1'
+version: 0.32.0
+appVersion: 'v0.39.2'
 
 dependencies:
   - name: crds

operations/helm/charts/grafana-agent/README.md

@@ -1,6 +1,6 @@
 # Grafana Agent Helm chart
 
-![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![Version: 0.31.1](https://img.shields.io/badge/Version-0.31.1-informational?style=flat-square) ![AppVersion: v0.39.1](https://img.shields.io/badge/AppVersion-v0.39.1-informational?style=flat-square)
+![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![Version: 0.32.0](https://img.shields.io/badge/Version-0.32.0-informational?style=flat-square) ![AppVersion: v0.39.2](https://img.shields.io/badge/AppVersion-v0.39.2-informational?style=flat-square)
 
 Helm chart for deploying [Grafana Agent][] to Kubernetes.
 
@@ -53,6 +53,7 @@ use the older mode (called "static mode"), set the `agent.mode` value to
 | agent.extraPorts | list | `[]` | Extra ports to expose on the Agent |
 | agent.listenAddr | string | `"0.0.0.0"` | Address to listen for traffic on. 0.0.0.0 exposes the UI to other containers. |
 | agent.listenPort | int | `80` | Port to listen for traffic on. |
+| agent.listenScheme | string | `"HTTP"` | Scheme is needed for readiness probes. If enabling tls in your configs, set to "HTTPS" |
 | agent.mode | string | `"flow"` | Mode to run Grafana Agent in. Can be "flow" or "static". |
 | agent.mounts.dockercontainers | bool | `false` | Mount /var/lib/docker/containers from the host into the container for log collection. |
 | agent.mounts.extra | list | `[]` | Extra volume mounts to add into the Grafana Agent container. Does not affect the watch container. |

operations/helm/charts/grafana-agent/templates/containers/_agent.yaml

@@ -55,6 +55,7 @@
     httpGet:
       path: /-/ready
       port: {{ .Values.agent.listenPort }}
+      scheme: {{ .Values.agent.listenScheme }}
     initialDelaySeconds: 10
     timeoutSeconds: 1
   {{- with .Values.agent.resources }}

operations/helm/charts/grafana-agent/values.yaml

@@ -53,6 +53,9 @@ agent:
   # -- Port to listen for traffic on.
   listenPort: 80
 
+  # -- Scheme is needed for readiness probes. If enabling tls in your configs, set to "HTTPS"
+  listenScheme: HTTP
+
   # --  Base path where the UI is exposed.
   uiPathPrefix: /
 

operations/helm/tests/additional-serviceaccount-label/grafana-agent/templates/controllers/daemonset.yaml

@@ -27,7 +27,7 @@ spec:
       serviceAccountName: grafana-agent
       containers:
         - name: grafana-agent
-          image: docker.io/grafana/agent:v0.39.1
+          image: docker.io/grafana/agent:v0.39.2
           imagePullPolicy: IfNotPresent
           args:
             - run
@@ -51,6 +51,7 @@ spec:
             httpGet:
               path: /-/ready
               port: 80
+              scheme: HTTP
             initialDelaySeconds: 10
             timeoutSeconds: 1
           volumeMounts:

operations/helm/tests/clustering/grafana-agent/templates/controllers/statefulset.yaml

@@ -30,7 +30,7 @@ spec:
       serviceAccountName: grafana-agent
       containers:
         - name: grafana-agent
-          image: docker.io/grafana/agent:v0.39.1
+          image: docker.io/grafana/agent:v0.39.2
           imagePullPolicy: IfNotPresent
           args:
             - run
@@ -56,6 +56,7 @@ spec:
             httpGet:
               path: /-/ready
               port: 80
+              scheme: HTTP
             initialDelaySeconds: 10
             timeoutSeconds: 1
           volumeMounts: