@@ -23,8 +23,10 @@ import (
2323 "errors"
2424 "fmt"
2525 "net/http"
26- "sync "
26+ "time "
2727
28+ "github.com/cespare/xxhash"
29+ "github.com/elastic/go-freelru"
2830 "github.com/elastic/opentelemetry-collector-components/extension/apmconfigextension/apmconfig"
2931 "github.com/open-telemetry/opamp-go/protobufs"
3032 "github.com/open-telemetry/opamp-go/server/types"
@@ -35,8 +37,10 @@ type remoteConfigCallbacks struct {
3537 * types.Callbacks
3638 configClient apmconfig.RemoteConfigClient
3739
38- agentState sync.Map
39- logger * zap.Logger
40+ agentState freelru.Cache [string , * agentInfo ]
41+ ttl time.Duration
42+
43+ logger * zap.Logger
4044}
4145
4246type agentInfo struct {
@@ -45,11 +49,35 @@ type agentInfo struct {
4549 lastConfigHash apmconfig.LastConfigHash
4650}
4751
48- func newRemoteConfigCallbacks (configClient apmconfig.RemoteConfigClient , logger * zap.Logger ) * remoteConfigCallbacks {
52+ func newRemoteConfigCallbacks (ctx context.Context , configClient apmconfig.RemoteConfigClient , ttlConfig CacheConfig , logger * zap.Logger ) (* remoteConfigCallbacks , error ) {
53+ cache , err := freelru .NewSharded [string , * agentInfo ](ttlConfig .Capacity , func (key string ) uint32 {
54+ return uint32 (xxhash .Sum64String (key ))
55+ })
56+ if err != nil {
57+ return nil , err
58+ }
59+ cache .SetLifetime (ttlConfig .TTL )
60+ // Purge expired entries from the cache
61+ if ttlConfig .TTL > 0 {
62+ go func () {
63+ ticker := time .NewTicker (ttlConfig .TTL )
64+ defer ticker .Stop ()
65+ for {
66+ select {
67+ case <- ctx .Done ():
68+ return
69+ case <- ticker .C :
70+ cache .PurgeExpired ()
71+ }
72+ }
73+ }()
74+ }
75+
4976 opampCallbacks := & remoteConfigCallbacks {
5077 configClient : configClient ,
51- agentState : sync. Map {} ,
78+ agentState : cache ,
5279 logger : logger ,
80+ ttl : ttlConfig .TTL ,
5381 }
5482
5583 connectionCallbacks := types.ConnectionCallbacks {}
@@ -67,7 +95,7 @@ func newRemoteConfigCallbacks(configClient apmconfig.RemoteConfigClient, logger
6795 },
6896 }
6997
70- return opampCallbacks
98+ return opampCallbacks , nil
7199}
72100
73101func (rc * remoteConfigCallbacks ) serverError (msg string , message * protobufs.ServerToAgent , logFields ... zap.Field ) * protobufs.ServerToAgent {
@@ -97,7 +125,7 @@ func (rc *remoteConfigCallbacks) onMessage(ctx context.Context, conn types.Conne
97125 agentUid := hex .EncodeToString (message .GetInstanceUid ())
98126 if message .GetAgentDescription () != nil {
99127 // new description might lead to another remote configuration
100- rc .agentState .Store (agentUid , agentInfo {
128+ _ = rc .agentState .Add (agentUid , & agentInfo {
101129 agentUid : message .GetInstanceUid (),
102130 identifyingAttributes : message .AgentDescription .IdentifyingAttributes ,
103131 })
@@ -106,26 +134,26 @@ func (rc *remoteConfigCallbacks) onMessage(ctx context.Context, conn types.Conne
106134 agentUidField := zap .String ("instance_uid" , agentUid )
107135 if message .GetAgentDisconnect () != nil {
108136 rc .logger .Info ("Disconnecting the agent from the remote configuration service" , agentUidField )
109- rc .agentState .Delete (agentUid )
137+ _ = rc .agentState .Remove (agentUid )
110138 return & serverToAgent
111139 }
112140
113- loadedAgent , _ := rc .agentState .LoadOrStore (agentUid , agentInfo {
114- agentUid : message .GetInstanceUid (),
115- })
116- agent , ok := loadedAgent .(agentInfo )
117- if ! ok {
118- rc .logger .Warn ("unexpected type in agentState cache" , agentUidField )
119- return rc .serverError ("internal error: invalid agent state" , & serverToAgent )
141+ loadedAgent , found := rc .agentState .GetAndRefresh (agentUid , rc .ttl )
142+ if ! found {
143+ loadedAgent = & agentInfo {
144+ agentUid : message .InstanceUid ,
145+ }
146+ _ = rc .agentState .Add (agentUid , loadedAgent )
120147 }
148+
121149 remoteConfigStatus := message .GetRemoteConfigStatus ()
122150 if remoteConfigStatus != nil {
123- agent .lastConfigHash = remoteConfigStatus .GetLastRemoteConfigHash ()
124- rc .logger .Info ("Remote config status" , agentUidField , zap .String ("lastRemoteConfigHash" , hex .EncodeToString (agent .lastConfigHash )), zap .String ("status" , remoteConfigStatus .GetStatus ().String ()), zap .String ("errorMessage" , remoteConfigStatus .ErrorMessage ))
125- rc .agentState .Store (agentUid , agent )
151+ loadedAgent .lastConfigHash = remoteConfigStatus .GetLastRemoteConfigHash ()
152+ rc .logger .Info ("Remote config status" , agentUidField , zap .String ("lastRemoteConfigHash" , hex .EncodeToString (loadedAgent .lastConfigHash )), zap .String ("status" , remoteConfigStatus .GetStatus ().String ()), zap .String ("errorMessage" , remoteConfigStatus .ErrorMessage ))
153+ rc .agentState .Add (agentUid , loadedAgent )
126154 }
127155
128- remoteConfig , err := rc .configClient .RemoteConfig (ctx , agent .identifyingAttributes , agent .lastConfigHash )
156+ remoteConfig , err := rc .configClient .RemoteConfig (ctx , loadedAgent .identifyingAttributes , loadedAgent .lastConfigHash )
129157 if err != nil {
130158 // remote config client could not identify the agent
131159 if errors .Is (err , apmconfig .UnidentifiedAgent ) {
0 commit comments