SH-455 refactoring legacy tags

Author: Rustamchuk

cmd/statshouse-api/statshouse-api.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/VKCOM/statshouse-go"
+	"github.com/VKCOM/statshouse/internal/vkgo/srvfunc"
 	"github.com/cloudflare/tableflip"
 	"github.com/gorilla/handlers"
 
@@ -35,7 +36,6 @@ import (
 	"github.com/VKCOM/statshouse/internal/util"
 	"github.com/VKCOM/statshouse/internal/vkgo/build"
 	"github.com/VKCOM/statshouse/internal/vkgo/rpc"
-	"github.com/VKCOM/statshouse/internal/vkgo/srvfunc"
 	"github.com/VKCOM/statshouse/internal/vkgo/vkuth"
 )
 
@@ -397,6 +397,7 @@ func run() int {
 	heartbeatTags := statshouse.Tags{
 		1: "4",
 		2: fmt.Sprint(format.TagValueIDHeartbeatEventStart),
+		4: fmt.Sprint(build.CommitTag()),
 		6: fmt.Sprint(build.CommitTimestamp()),
 		7: srvfunc.HostnameForStatshouse(),
 	}

cmd/statshouse-igp/ingress_proxy.go

@@ -594,6 +594,7 @@ func (p *proxyConn) reportRequestSize(req *proxyRequest) {
 			1: format.TagValueIDComponentIngressProxy,
 			2: int32(req.tag()),
 			6: p.clientCryptoKeyID,
+			7: p.hostnameID.Load(),
 			8: p.clientProtocolVersion,
 		}, float64(req.size), 1)
 }

internal/agent/agent.go

@@ -662,13 +662,38 @@ func (s *Agent) addBuiltins(nowUnix uint32) {
 }
 
 func (s *Agent) addBuiltInsHeartbeatsLocked(nowUnix uint32, count float64) {
-	uptimeSec := float64(nowUnix - s.startTimestamp)
-
-	s.AddValueCounterString(nowUnix, &s.builtinMetricMetaHeartbeatVersion,
-		[]int32{0, s.componentTag, s.heartBeatEventType},
+	var (
+		owner     string
+		uptimeSec = float64(nowUnix - s.startTimestamp)
+	)
+	if s.envLoader != nil {
+		e := s.envLoader.Load()
+		owner = e.Owner
+	}
+	s.AddValueCounterStringV2(nowUnix, &s.builtinMetricMetaHeartbeatVersion,
+		[]int32{
+			1: s.componentTag,
+			2: s.heartBeatEventType,
+			4: int32(build.CommitTag()),
+			6: int32(build.CommitTimestamp()),
+		},
+		[]string{
+			7: string(s.hostName),
+			9: owner,
+		},
 		build.Commit(), uptimeSec, count)
-	s.AddValueCounterString(nowUnix, &s.builtinMetricMetaHeartbeatArgs,
-		[]int32{0, s.componentTag, s.heartBeatEventType, s.argsHash, 0, 0, 0, 0, 0, s.argsLen},
+	s.AddValueCounterStringV2(nowUnix, &s.builtinMetricMetaHeartbeatArgs,
+		[]int32{
+			1: s.componentTag,
+			2: s.heartBeatEventType,
+			3: s.argsHash,
+			4: int32(build.CommitTag()),
+			6: int32(build.CommitTimestamp()),
+		},
+		[]string{
+			7: string(s.hostName),
+			9: owner,
+		},
 		s.args, uptimeSec, count)
 }
 
@@ -945,6 +970,33 @@ func (s *Agent) AddValueCounterStringHostAERA(t uint32, metricInfo *format.Metri
 	shard.AddValueCounterStringHost(&key, resolutionHash, topValue, value, counter, hostTag, metricInfo, weightMul)
 }
 
+func (s *Agent) AddValueCounterStringV2(t uint32, metricInfo *format.MetricMetaValue, tags []int32, stags []string, str string, value float64, counter float64) {
+	s.AddValueCounterStringV2HostAERA(t, metricInfo, tags, stags, data_model.TagUnion{S: str, I: 0}, value, counter, data_model.TagUnionBytes{}, format.AgentEnvRouteArch{})
+}
+
+func (s *Agent) AddValueCounterStringV2HostAERA(t uint32, metricInfo *format.MetricMetaValue, tags []int32, stags []string, topValue data_model.TagUnion, value float64, counter float64, hostTag data_model.TagUnionBytes, aera format.AgentEnvRouteArch) {
+	if counter <= 0 {
+		return
+	}
+	key := data_model.Key{Timestamp: t, Metric: metricInfo.MetricID} // panics if metricInfo nil
+	copy(key.Tags[:], tags)
+	copy(key.STags[:], stags)
+	if metricInfo.WithAggregatorID {
+		key.Tags[format.AggHostTag] = s.AggregatorHost
+		key.Tags[format.AggShardTag] = s.AggregatorShardKey
+		key.Tags[format.AggReplicaTag] = s.AggregatorReplicaKey
+	}
+	if metricInfo.WithAgentEnvRouteArch {
+		key.Tags[format.AgentEnvTag] = aera.AgentEnv
+		key.Tags[format.RouteTag] = aera.Route
+		key.Tags[format.BuildArchTag] = aera.BuildArch
+	}
+	shardId, _, weightMul, resolutionHash := s.shard(&key, metricInfo, nil)
+	// resolutionHash will be 0 for built-in metrics, we are OK with this
+	shard := s.Shards[shardId]
+	shard.AddValueCounterStringHost(&key, resolutionHash, topValue, value, counter, hostTag, metricInfo, weightMul)
+}
+
 func (s *Agent) MergeItemValue(t uint32, metricInfo *format.MetricMetaValue, tags []int32, item *data_model.ItemValue) {
 	if item.Count() <= 0 {
 		return

internal/aggregator/aggregator_handlers.go

@@ -50,8 +50,14 @@ func (a *Aggregator) handleClient(ctx context.Context, hctx *rpc.HandlerContext)
 		str = err.Error()
 	}
 	a.sh2.AddValueCounterString(uint32(hctx.RequestTime.Unix()), format.BuiltinMetricMetaRPCRequests,
-		[]int32{0, format.TagValueIDComponentAggregator, int32(tag), status, 0, 0, keyIDTag, 0, protocol},
-		str, float64(requestLen), 1)
+		[]int32{
+			1: format.TagValueIDComponentAggregator,
+			2: int32(tag),
+			3: status,
+			6: keyIDTag,
+			7: a.aggregatorHost,
+			8: protocol,
+		}, str, float64(requestLen), 1)
 	return err
 }
 
@@ -502,17 +508,12 @@ func (a *Aggregator) handleSendSourceBucketAny(hctx *rpc.HandlerContext, args tl
 				k.Tags[format.RouteTag] = aera.Route
 				k.Tags[format.BuildArchTag] = aera.BuildArch
 			}
+			if k.Metric == format.BuiltinMetricIDHeartbeatVersion || k.Metric == format.BuiltinMetricIDHeartbeatArgs {
+				k.Tags[8] = int32(addrIPV4)
+			}
+
+			// TODO - remove
 			switch k.Metric {
-			case format.BuiltinMetricIDAgentHeartbeatVersion:
-				// Remap legacy metric to a new one
-				k.Metric = format.BuiltinMetricIDHeartbeatVersion
-				k.Tags[2] = k.Tags[1]
-				k.Tags[1] = format.TagValueIDComponentAgent
-			case format.BuiltinMetricIDAgentHeartbeatArgs:
-				// Remap legacy metric to a new one
-				k.Metric = format.BuiltinMetricIDHeartbeatArgs
-				k.Tags[2] = k.Tags[1]
-				k.Tags[1] = format.TagValueIDComponentAgent
 			case format.BuiltinMetricIDHeartbeatVersion, format.BuiltinMetricIDHeartbeatArgs:
 				// In case of agent we need to set IP anyway, so set other keys here, not by source
 				// In case of api other tags are already set, so don't overwrite them

internal/format/builtin_metrics.go

@@ -806,9 +806,6 @@ var BuiltinMetricMetaPromScrapeTime = &MetricMetaValue{
 	},
 }
 
-const BuiltinMetricIDAgentHeartbeatVersion = -41 // TODO - remove
-const BuiltinMetricIDAgentHeartbeatArgs = -42    // TODO - remove, this metric was writing larger than allowed strings to DB in the past
-
 var BuiltinMetricMetaUsageMemory = &MetricMetaValue{
 	Name:                    "__usage_mem",
 	Kind:                    MetricKindValue,