cherry picks for 1.10.2 (grafana#4218)

* Update the walqueue to get fix for shard management (grafana#4211)

(cherry picked from commit 41b6ea3)

* fix: gcplog push message to support both message_id and messageId (grafana#4110)

* Support both variants for push messages from gcp

(cherry picked from commit 853309c)

* Update changelog

* Fix changelog

---------

Co-authored-by: Karl Persson <23356117+kalleep@users.noreply.github.com>

Author: dehaansa

CHANGELOG.md

@@ -7,6 +7,15 @@ This document contains a historical list of changes between releases. Only
 changes that impact end-user behavior are listed; changes to documentation or
 internal API changes are not present.
 
+v1.10.2
+-----------------
+
+### Bugfixes
+
+- Fix issue in `prometheus.write.queue` causing inability to increase shard count if existing WAL data was present on start. (@kgeckhart)
+
+- Fix issue with `loki.source.gcplog` when push messages sent by gcp pub/sub only includes `messageId`. (@kalleep)
+
 v1.10.1
 -----------------
 

go.mod

@@ -76,7 +76,7 @@ require (
 	github.com/grafana/snowflake-prometheus-exporter v0.0.0-20250627131542-0c2feac3a700
 	github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0
 	github.com/grafana/vmware_exporter v0.0.5-beta.0.20250218170317-73398ba08329
-	github.com/grafana/walqueue v0.0.0-20250806200024-8857bde78364 // alloy-1.10 branch to cherry pick fixes w/o prom3 update
+	github.com/grafana/walqueue v0.0.0-20250818155145-61b631a3f799 // alloy-1.10 branch to cherry pick fixes w/o prom3 update
 	github.com/hashicorp/consul/api v1.32.1
 	github.com/hashicorp/go-discover v1.1.0
 	github.com/hashicorp/go-multierror v1.1.1
@@ -528,7 +528,7 @@ require (
 	github.com/cyphar/filepath-securejoin v0.4.1 // indirect
 	github.com/danieljoos/wincred v1.2.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/deneonet/benc v1.1.8 // indirect
+	github.com/deneonet/benc v1.1.7 // indirect
 	github.com/dennwc/btrfs v0.0.0-20230312211831-a1f570bd01a1 // indirect
 	github.com/dennwc/ioctl v1.0.0 // indirect
 	github.com/dennwc/varint v1.0.0 // indirect
@@ -1102,3 +1102,6 @@ replace github.com/prometheus/procfs => github.com/prometheus/procfs v0.12.0
 
 // Use v0.62.0 of prometheus/common for all dependencies until mongodb_exporter is updated to support 0.63.0
 replace github.com/prometheus/common => github.com/prometheus/common v0.62.0
+
+// Do not remove this until the bug breaking backwards compatibility is resolved & included in walqueue: https://github.com/deneonet/benc/issues/13
+replace github.com/deneonet/benc => github.com/deneonet/benc v1.1.7

go.sum

@@ -872,8 +872,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/deneonet/benc v1.1.8 h1:Qk9diyH0UcnduvCrZ62mBrwUeSZzte4kQxMbclVdhW4=
-github.com/deneonet/benc v1.1.8/go.mod h1:UCfkM5Od0B2huwv/ZItvtUb7QnALFt9YXtX8NXX4Lts=
+github.com/deneonet/benc v1.1.7 h1:0XPxTTVJZq/ulxXvMn2Mzjx5XquekVky3wX6eTgA0vA=
+github.com/deneonet/benc v1.1.7/go.mod h1:UCfkM5Od0B2huwv/ZItvtUb7QnALFt9YXtX8NXX4Lts=
 github.com/denisenkom/go-mssqldb v0.0.0-20180620032804-94c9c97e8c9f/go.mod h1:xN/JuLBIz4bjkxNmByTiV1IbhfnYb6oo99phBn4Eqhc=
 github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4/go.mod h1:zAg7JM8CkOJ43xKXIj7eRO9kmWm/TW578qo+oDO6tuM=
 github.com/denisenkom/go-mssqldb v0.0.0-20191128021309-1d7a30a10f73/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
@@ -1449,8 +1449,8 @@ github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0 h1:bjh0PVYSVVFxzINqPF
 github.com/grafana/tail v0.0.0-20230510142333-77b18831edf0/go.mod h1:7t5XR+2IA8P2qggOAHTj/GCZfoLBle3OvNSYh1VkRBU=
 github.com/grafana/vmware_exporter v0.0.5-beta.0.20250218170317-73398ba08329 h1:Rs4H1yv2Abk3xE82qpyhMGGA8rswAOA0HQZde/BYkFo=
 github.com/grafana/vmware_exporter v0.0.5-beta.0.20250218170317-73398ba08329/go.mod h1:Z28219aViNlsLlPvuCnlgHDagRdZBAZ7JOnQg1b3eWg=
-github.com/grafana/walqueue v0.0.0-20250806200024-8857bde78364 h1:KNk1ugoycr2YBY+jISAUWSl8u7u0DBMFzBTBmLl3juI=
-github.com/grafana/walqueue v0.0.0-20250806200024-8857bde78364/go.mod h1:r5esBN3LfoYHk5JKD+o4/IBwuftA1uCmaqT6C8av21c=
+github.com/grafana/walqueue v0.0.0-20250818155145-61b631a3f799 h1:PtCAPUoGn0FLxBwAs7AsV9Lse9ayQSzYMDUxQp1yYOM=
+github.com/grafana/walqueue v0.0.0-20250818155145-61b631a3f799/go.mod h1:VBs5QI+CljQRsh+x7K8QamyDQ/nTOPqCRdk8h31Kb7Y=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grobie/gomemcache v0.0.0-20230213081705-239240bbc445 h1:FlKQKUYPZ5yDCN248M3R7x8yu2E3yEZ0H7aLomE4EoE=
 github.com/grobie/gomemcache v0.0.0-20230213081705-239240bbc445/go.mod h1:L69/dBlPQlWkcnU76WgcppK5e4rrxzQdi6LhLnK/ytA=

internal/component/loki/source/gcplog/internal/gcplogtarget/push_target.go

@@ -76,7 +76,7 @@ func (p *PushTarget) push(w http.ResponseWriter, r *http.Request) {
 	}
 	defer cancel()
 
-	pushMessage := PushMessage{}
+	pushMessage := PushMessageBody{}
 	bs, err := io.ReadAll(r.Body)
 	if err != nil {
 		p.metrics.gcpPushErrors.WithLabelValues("read_error").Inc()

internal/component/loki/source/gcplog/internal/gcplogtarget/push_translation.go

@@ -22,24 +22,26 @@ import (
 // pipeline stages
 const ReservedLabelTenantID = "__tenant_id__"
 
-// PushMessage is the POST body format sent by GCP PubSub push subscriptions.
+// PushMessageBody is the POST body format sent by GCP PubSub push subscriptions.
 // See https://cloud.google.com/pubsub/docs/push for details.
+type PushMessageBody struct {
+	Message      PushMessage `json:"message"`
+	Subscription string      `json:"subscription"`
+}
+
 type PushMessage struct {
-	Message struct {
-		Attributes       map[string]string `json:"attributes"`
-		Data             string            `json:"data"`
-		ID               string            `json:"message_id"`
-		PublishTimestamp string            `json:"publish_time"`
-	} `json:"message"`
-	Subscription string `json:"subscription"`
+	Attributes          map[string]string `json:"attributes"`
+	Data                string            `json:"data"`
+	MessageID           string            `json:"messageId"`
+	DeprecatedMessageID string            `json:"message_id"`
 }
 
 // Validate checks that the required fields of a PushMessage are set.
-func (pm PushMessage) Validate() error {
+func (pm PushMessageBody) Validate() error {
 	if pm.Message.Data == "" {
 		return fmt.Errorf("push message has no data")
 	}
-	if pm.Message.ID == "" {
+	if pm.ID() == "" {
 		return fmt.Errorf("push message has no ID")
 	}
 	if pm.Subscription == "" {
@@ -48,14 +50,21 @@ func (pm PushMessage) Validate() error {
 	return nil
 }
 
+func (pm PushMessageBody) ID() string {
+	if pm.Message.MessageID != "" {
+		return pm.Message.MessageID
+	}
+	return pm.Message.DeprecatedMessageID
+}
+
 // translate converts a GCP PushMessage into a loki.Entry. It parses the
 // push-specific labels and delegates the rest to parseGCPLogsEntry.
-func translate(m PushMessage, other model.LabelSet, useIncomingTimestamp bool, useFullLine bool, relabelConfigs []*relabel.Config, xScopeOrgID string) (loki.Entry, error) {
+func translate(m PushMessageBody, other model.LabelSet, useIncomingTimestamp bool, useFullLine bool, relabelConfigs []*relabel.Config, xScopeOrgID string) (loki.Entry, error) {
 	// Collect all push-specific labels. Every one of them is first configured
 	// as optional, and the user can relabel it if needed. The relabeling and
 	// internal drop is handled in parseGCPLogsEntry.
 	lbs := labels.NewBuilder(nil)
-	lbs.Set("__gcp_message_id", m.Message.ID)
+	lbs.Set("__gcp_message_id", m.ID())
 	lbs.Set("__gcp_subscription_name", m.Subscription)
 	for k, v := range m.Message.Attributes {
 		lbs.Set(fmt.Sprintf("__gcp_attributes_%s", convertToLokiCompatibleLabel(k)), v)

internal/component/loki/source/gcplog/internal/gcplogtarget/push_translation_test.go

@@ -2,6 +2,13 @@ package gcplogtarget
 
 import (
 	"testing"
+
+	"github.com/grafana/regexp"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/relabel"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/alloy/internal/component/common/loki"
 )
 
 func TestConvertToLokiCompatibleLabel(t *testing.T) {
@@ -37,9 +44,76 @@ func TestConvertToLokiCompatibleLabel(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := convertToLokiCompatibleLabel(tt.args.label); got != tt.want {
-				t.Errorf("convertToLokiCompatibleLabel() = %v, want %v", got, tt.want)
-			}
+			require.Equal(t, tt.want, convertToLokiCompatibleLabel(tt.args.label))
+		})
+	}
+}
+
+const pushData = "eyJpbnNlcnRJZCI6IjRhZmZhODU4LWU1ZjItNDdmNy05MjU0LWU2MDliNWMwMTRkMCIsImxhYmVscyI6e30sImxvZ05hbWUiOiJwcm9qZWN0cy90ZXN0LXByb2plY3QvbG9ncy9jbG91ZGF1ZGl0Lmdvb2dsZWFwaXMuY29tJTJGZGF0YV9hY2Nlc3MiLCJyZWNlaXZlVGltZXN0YW1wIjoiMjAyMi0wOS0wNlQxODowNzo0My40MTc3MTQwNDZaIiwicmVzb3VyY2UiOnsibGFiZWxzIjp7ImNsdXN0ZXJfbmFtZSI6ImRldi11cy1jZW50cmFsLTQyIiwibG9jYXRpb24iOiJ1cy1jZW50cmFsMSIsInByb2plY3RfaWQiOiJ0ZXN0LXByb2plY3QifSwidHlwZSI6Ims4c19jbHVzdGVyIn0sInRpbWVzdGFtcCI6IjIwMjItMDktMDZUMTg6MDc6NDIuMzYzMTEzWiJ9Cg=="
+
+func TestTranslate(t *testing.T) {
+	type testCase struct {
+		name     string
+		pm       PushMessageBody
+		expected loki.Entry
+	}
+
+	tests := []testCase{
+		{
+			name: "deprecated message id",
+			pm: PushMessageBody{
+				Message: PushMessage{
+					DeprecatedMessageID: "1",
+					Data:                pushData,
+				},
+				Subscription: "test",
+			},
+			expected: loki.Entry{
+				Labels: model.LabelSet{
+					"message_id": "1",
+				},
+			},
+		},
+		{
+			name: "standard message id",
+			pm: PushMessageBody{
+				Message: PushMessage{
+					MessageID: "1",
+					Data:      pushData,
+				},
+				Subscription: "test",
+			},
+			expected: loki.Entry{
+				Labels: model.LabelSet{
+					"message_id": "1",
+				},
+			},
+		},
+	}
+
+	rc := []*relabel.Config{
+		{
+			SourceLabels: model.LabelNames{"__gcp_message_id"},
+			Regex:        mustNewRegexp("(.*)"),
+			Action:       relabel.Replace,
+			Replacement:  "$1",
+			TargetLabel:  "message_id",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			entry, err := translate(tt.pm, model.LabelSet{}, false, false, rc, "")
+			require.NoError(t, err)
+			require.EqualValues(t, tt.expected.Labels, entry.Labels)
 		})
 	}
 }
+
+func mustNewRegexp(s string) relabel.Regexp {
+	re, err := regexp.Compile("^(?:" + s + ")$")
+	if err != nil {
+		panic(err)
+	}
+	return relabel.Regexp{Regexp: re}
+}