diff --git a/controller/env/appenv.go b/controller/env/appenv.go index 0ebf81a77..289216c6c 100644 --- a/controller/env/appenv.go +++ b/controller/env/appenv.go @@ -53,7 +53,7 @@ import ( "github.com/openziti/sdk-golang/ziti/constants" "github.com/openziti/storage/boltz" "github.com/openziti/xweb" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/xeipuuv/gojsonschema" "io" "io/ioutil" @@ -83,7 +83,7 @@ type AppEnv struct { HostController HostController ManagementApi *managementOperations.ZitiEdgeManagementAPI ClientApi *clientOperations.ZitiEdgeClientAPI - IdentityRefreshMap cmap.ConcurrentMap + IdentityRefreshMap cmap.ConcurrentMap[time.Time] identityRefreshMeter metrics.Meter StartupTime time.Time InstanceId string @@ -374,7 +374,7 @@ func NewAppEnv(c *edgeConfig.Config, host HostController) *AppEnv { EnrollRegistry: &model.EnrollmentRegistryImpl{}, ManagementApi: managementApi, ClientApi: clientApi, - IdentityRefreshMap: cmap.New(), + IdentityRefreshMap: cmap.New[time.Time](), StartupTime: time.Now().UTC(), } diff --git a/controller/env/trace.go b/controller/env/trace.go index f2f72a349..083038a9c 100644 --- a/controller/env/trace.go +++ b/controller/env/trace.go @@ -2,13 +2,13 @@ package env import ( "fmt" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "time" ) func NewTraceManager(shutdownNotify <-chan struct{}) *TraceManager { result := &TraceManager{ - traceIdentities: cmap.New(), + traceIdentities: cmap.New[*TraceSpec](), shutdownNotify: shutdownNotify, } go result.reapExpired() @@ -26,16 +26,17 @@ func (self *TraceSpec) String() string { } type TraceManager struct { - traceIdentities cmap.ConcurrentMap + traceIdentities cmap.ConcurrentMap[*TraceSpec] shutdownNotify <-chan struct{} } func (self *TraceManager) GetIdentityTrace(identityId string) *TraceSpec { - val, found := self.traceIdentities.Get(identityId) + spec, found := self.traceIdentities.Get(identityId) + if !found { return nil } - spec := val.(*TraceSpec) + specCopy := *spec return &specCopy } @@ -64,8 +65,7 @@ func (self *TraceManager) reapExpired() { case <-ticker.C: now := time.Now() var toRemove []string - self.traceIdentities.IterCb(func(key string, v interface{}) { - spec := v.(*TraceSpec) + self.traceIdentities.IterCb(func(key string, spec *TraceSpec) { if spec.Until.Before(now) { toRemove = append(toRemove, key) } diff --git a/controller/handler_edge_ctrl/list_tunnel_services.go b/controller/handler_edge_ctrl/list_tunnel_services.go index 12ca56f57..53243848d 100644 --- a/controller/handler_edge_ctrl/list_tunnel_services.go +++ b/controller/handler_edge_ctrl/list_tunnel_services.go @@ -2,12 +2,12 @@ package handler_edge_ctrl import ( "encoding/json" - "google.golang.org/protobuf/proto" "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/storage/ast" "github.com/sirupsen/logrus" + "google.golang.org/protobuf/proto" "time" ) @@ -76,7 +76,7 @@ func (self *listTunnelServicesHandler) listServices(ctx *listTunnelServicesReque var lastUpdate time.Time if val, found := self.appEnv.IdentityRefreshMap.Get(ctx.identity.Id); found { - lastUpdate = val.(time.Time) + lastUpdate = val } else { lastUpdate = self.appEnv.StartupTime } diff --git a/controller/internal/routes/current_api_session_router.go b/controller/internal/routes/current_api_session_router.go index daa2b04da..446421f3c 100644 --- a/controller/internal/routes/current_api_session_router.go +++ b/controller/internal/routes/current_api_session_router.go @@ -195,7 +195,7 @@ func (router *CurrentSessionRouter) DeleteCertificate(ae *env.AppEnv, rc *respon func (router *CurrentSessionRouter) ListServiceUpdates(ae *env.AppEnv, rc *response.RequestContext) { lastUpdate := rc.ApiSession.CreatedAt if val, found := ae.IdentityRefreshMap.Get(rc.Identity.Id); found { - lastUpdate = val.(time.Time) + lastUpdate = val } else if lastUpdate.Before(ae.StartupTime) { lastUpdate = ae.StartupTime } diff --git a/controller/model/api_session_heartbeats.go b/controller/model/api_session_heartbeats.go index f13535475..8f99829f7 100644 --- a/controller/model/api_session_heartbeats.go +++ b/controller/model/api_session_heartbeats.go @@ -19,14 +19,14 @@ package model import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/controller/persistence" - "github.com/openziti/storage/boltz" "github.com/openziti/foundation/util/concurrenz" - cmap "github.com/orcaman/concurrent-map" + "github.com/openziti/storage/boltz" + cmap "github.com/orcaman/concurrent-map/v2" "time" ) type HeartbeatCollector struct { - apiSessionLastAccessedAtMap cmap.ConcurrentMap //map[apiSessionId string] => *HeartbeatStatus + apiSessionLastAccessedAtMap cmap.ConcurrentMap[*HeartbeatStatus] updateInterval time.Duration closeNotify chan struct{} isFlushing concurrenz.AtomicBoolean @@ -45,7 +45,7 @@ type Heartbeat struct { // buffer for last update times. func NewHeartbeatCollector(env Env, batchSize int, updateInterval time.Duration, action func([]*Heartbeat)) *HeartbeatCollector { collector := &HeartbeatCollector{ - apiSessionLastAccessedAtMap: cmap.New(), + apiSessionLastAccessedAtMap: cmap.New[*HeartbeatStatus](), updateInterval: updateInterval, batchSize: batchSize, flushAction: action, @@ -76,11 +76,9 @@ func (self *HeartbeatCollector) Mark(apiSessionId string) { // or made a REST API call and true. If no such action has happened or the API Session no longer exists // nil and false will be returned. func (self *HeartbeatCollector) LastAccessedAt(apiSessionId string) (*time.Time, bool) { - if val, ok := self.apiSessionLastAccessedAtMap.Get(apiSessionId); ok { - if status, ok := val.(*HeartbeatStatus); ok { - lastAccessedAt := status.lastAccessedAt - return &lastAccessedAt, true - } + if status, ok := self.apiSessionLastAccessedAtMap.Get(apiSessionId); ok { + lastAccessedAt := status.lastAccessedAt + return &lastAccessedAt, true } return nil, false @@ -114,20 +112,18 @@ func (self *HeartbeatCollector) flush() { var beats []*Heartbeat - self.apiSessionLastAccessedAtMap.IterCb(func(key string, v interface{}) { + self.apiSessionLastAccessedAtMap.IterCb(func(key string, status *HeartbeatStatus) { if len(beats) >= self.batchSize { self.flushAction(beats) beats = nil } - if status, ok := v.(*HeartbeatStatus); ok { - if !status.flushed { - status.flushed = true - beats = append(beats, &Heartbeat{ - ApiSessionId: key, - LastActivityAt: status.lastAccessedAt, - }) - } + if !status.flushed { + status.flushed = true + beats = append(beats, &Heartbeat{ + ApiSessionId: key, + LastActivityAt: status.lastAccessedAt, + }) } }) diff --git a/controller/model/authenticator_mod_cert.go b/controller/model/authenticator_mod_cert.go index b3c44c327..1ba7248ce 100644 --- a/controller/model/authenticator_mod_cert.go +++ b/controller/model/authenticator_mod_cert.go @@ -28,7 +28,7 @@ import ( "github.com/openziti/foundation/util/errorz" nfpem "github.com/openziti/foundation/util/pem" "github.com/openziti/foundation/util/stringz" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "net/http" "time" ) @@ -46,7 +46,7 @@ type AuthModuleCert struct { fingerprintGenerator cert.FingerprintGenerator caChain []byte staticCaCerts []*x509.Certificate - dynamicCaCache cmap.ConcurrentMap //map[string][]*x509.Certificate + dynamicCaCache cmap.ConcurrentMap[[]*x509.Certificate] } func NewAuthModuleCert(env Env, caChain []byte) *AuthModuleCert { @@ -55,7 +55,7 @@ func NewAuthModuleCert(env Env, caChain []byte) *AuthModuleCert { method: persistence.MethodAuthenticatorCert, fingerprintGenerator: cert.NewFingerprintGenerator(), staticCaCerts: nfpem.PemBytesToCertificates(caChain), - dynamicCaCache: cmap.New(), + dynamicCaCache: cmap.New[[]*x509.Certificate](), } return handler @@ -263,13 +263,11 @@ func (module *AuthModuleCert) getCas() *caPool { return nil } - if val, ok := module.dynamicCaCache.Get(ca.Id); ok { - if caCerts, ok := val.([]*x509.Certificate); ok { - for _, caCert := range caCerts { - result.roots.AddCert(caCert) - fingerprint := module.env.GetFingerprintGenerator().FromCert(caCert) - result.cas[fingerprint] = ca - } + if caCerts, ok := module.dynamicCaCache.Get(ca.Id); ok { + for _, caCert := range caCerts { + result.roots.AddCert(caCert) + fingerprint := module.env.GetFingerprintGenerator().FromCert(caCert) + result.cas[fingerprint] = ca } } else { caCerts := nfpem.PemStringToCertificates(ca.CertPem) diff --git a/controller/model/authenticator_mod_ext_jwt.go b/controller/model/authenticator_mod_ext_jwt.go index c8a4901a6..8fc51bfad 100644 --- a/controller/model/authenticator_mod_ext_jwt.go +++ b/controller/model/authenticator_mod_ext_jwt.go @@ -23,7 +23,7 @@ import ( "github.com/openziti/edge/controller/persistence" nfPem "github.com/openziti/foundation/util/pem" "github.com/openziti/storage/boltz" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "go.etcd.io/bbolt" "strings" ) @@ -38,14 +38,14 @@ const ( type AuthModuleExtJwt struct { env Env method string - signers cmap.ConcurrentMap //map[kid string]*signer + signers cmap.ConcurrentMap[*signerRecord] } func NewAuthModuleExtJwt(env Env) *AuthModuleExtJwt { ret := &AuthModuleExtJwt{ env: env, method: AuthMethodExtJwt, - signers: cmap.New(), + signers: cmap.New[*signerRecord](), } env.GetStores().ExternalJwtSigner.AddListener(boltz.EventCreate, ret.onExternalSignerCreate) @@ -66,8 +66,6 @@ func (a *AuthModuleExtJwt) CanHandle(method string) bool { return method == a.method } func (a *AuthModuleExtJwt) pubKeyLookup(token *jwt.Token) (interface{}, error) { - kidToSignerRectInterface := a.getKnownSignerRecords() - kidVal, ok := token.Header["kid"] if !ok { @@ -82,15 +80,13 @@ func (a *AuthModuleExtJwt) pubKeyLookup(token *jwt.Token) (interface{}, error) { return nil, apierror.NewInvalidAuth() } - signerRecordInterface, ok := kidToSignerRectInterface[kid] + signerRecord, ok := a.signers.Get(kid) if !ok { pfxlog.Logger().Error("unknown kid") return nil, apierror.NewInvalidAuth() } - signerRecord := signerRecordInterface.(*signerRecord) - if !signerRecord.externalJwtSigner.Enabled { pfxlog.Logger().WithField("externalJwtId", signerRecord.externalJwtSigner.Id).Error("external jwt is disabled") return nil, apierror.NewInvalidAuth() @@ -316,10 +312,6 @@ func (a *AuthModuleExtJwt) process(context AuthContext, isPrimary bool) (AuthRes return nil, apierror.NewInvalidAuth() } -func (a *AuthModuleExtJwt) getKnownSignerRecords() map[string]interface{} { - return a.signers.Items() -} - func (a *AuthModuleExtJwt) onExternalSignerCreate(args ...interface{}) { signer, ok := args[0].(*persistence.ExternalJwtSigner) diff --git a/controller/model/authenticator_mod_updb.go b/controller/model/authenticator_mod_updb.go index 3b6e63c6f..e40fc881b 100644 --- a/controller/model/authenticator_mod_updb.go +++ b/controller/model/authenticator_mod_updb.go @@ -23,7 +23,7 @@ import ( "github.com/openziti/edge/controller/apierror" "github.com/openziti/edge/controller/persistence" "github.com/openziti/foundation/util/errorz" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "time" ) @@ -32,14 +32,14 @@ var _ AuthProcessor = &AuthModuleUpdb{} type AuthModuleUpdb struct { env Env method string - attemptsByAuthenticatorId cmap.ConcurrentMap //string -> int64 + attemptsByAuthenticatorId cmap.ConcurrentMap[int64] } func NewAuthModuleUpdb(env Env) *AuthModuleUpdb { handler := &AuthModuleUpdb{ env: env, method: "password", - attemptsByAuthenticatorId: cmap.New(), + attemptsByAuthenticatorId: cmap.New[int64](), } return handler @@ -113,14 +113,13 @@ func (handler *AuthModuleUpdb) Process(context AuthContext) (AuthResult, error) } attempts := int64(0) - handler.attemptsByAuthenticatorId.Upsert(authenticator.Id, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + handler.attemptsByAuthenticatorId.Upsert(authenticator.Id, 0, func(exist bool, prevAttempts int64, newValue int64) int64 { if exist { - prevAttempts := valueInMap.(int64) attempts = prevAttempts + 1 return attempts } - return int64(0) + return 0 }) if authPolicy.Primary.Updb.MaxAttempts != persistence.UpdbUnlimitedAttemptsLimit && attempts > authPolicy.Primary.Updb.MaxAttempts { diff --git a/controller/model/identity_handlers.go b/controller/model/identity_handlers.go index 64c626830..0837710c0 100644 --- a/controller/model/identity_handlers.go +++ b/controller/model/identity_handlers.go @@ -24,9 +24,9 @@ import ( "github.com/openziti/edge/eid" "github.com/openziti/fabric/controller/models" "github.com/openziti/foundation/metrics" - "github.com/openziti/storage/boltz" "github.com/openziti/foundation/util/errorz" - cmap "github.com/orcaman/concurrent-map" + "github.com/openziti/storage/boltz" + cmap "github.com/orcaman/concurrent-map/v2" "go.etcd.io/bbolt" "sync" "time" @@ -573,9 +573,9 @@ func (handler *IdentityHandler) Enable(identityId string) error { } type identityStatusMap struct { - identities cmap.ConcurrentMap // identityId -> status{expiresAt} - initOnce sync.Once - activeDuration time.Duration + identityIdToStatus cmap.ConcurrentMap[*status] + initOnce sync.Once + activeDuration time.Duration } type status struct { @@ -584,24 +584,22 @@ type status struct { func newIdentityStatusMap(activeDuration time.Duration) *identityStatusMap { return &identityStatusMap{ - identities: cmap.New(), - activeDuration: activeDuration, + identityIdToStatus: cmap.New[*status](), + activeDuration: activeDuration, } } func (statusMap *identityStatusMap) SetActive(identityId string) { statusMap.initOnce.Do(statusMap.start) - statusMap.identities.Set(identityId, &status{ + statusMap.identityIdToStatus.Set(identityId, &status{ expiresAt: time.Now().Add(statusMap.activeDuration), }) } func (statusMap *identityStatusMap) IsActive(identityId string) bool { - if val, ok := statusMap.identities.Get(identityId); ok { - if status, ok := val.(*status); ok { - return status.expiresAt.After(time.Now()) - } + if stat, ok := statusMap.identityIdToStatus.Get(identityId); ok { + return stat.expiresAt.After(time.Now()) } return false } @@ -614,14 +612,14 @@ func (statusMap *identityStatusMap) start() { case <-ticker.C: var toRemove []string now := time.Now() - statusMap.identities.IterCb(func(key string, val interface{}) { - if status, ok := val.(*status); !ok || status.expiresAt.Before(now) { + statusMap.identityIdToStatus.IterCb(func(key string, stat *status) { + if stat.expiresAt.Before(now) { toRemove = append(toRemove, key) } }) for _, identityId := range toRemove { - statusMap.identities.Remove(identityId) + statusMap.identityIdToStatus.Remove(identityId) } } } diff --git a/controller/model/posture_response_handlers.go b/controller/model/posture_response_handlers.go index 56df6250d..7bc2b89c2 100644 --- a/controller/model/posture_response_handlers.go +++ b/controller/model/posture_response_handlers.go @@ -87,13 +87,13 @@ func (handler *PostureResponseHandler) SetMfaPostureForIdentity(identityId strin passedAt = &postureResponse.LastUpdatedAt } - handler.postureCache.Upsert(identityId, true, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + handler.postureCache.Upsert(identityId, true, func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData { var pd *PostureData if exist { - pd = valueInMap.(*PostureData) + pd = valueInMap } else { - pd = newValue.(*PostureData) + pd = newValue } if pd == nil { @@ -232,12 +232,12 @@ func (handler *PostureResponseHandler) SetSdkInfo(identityId, apiSessionId strin return } - handler.postureCache.Upsert(identityId, false, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + handler.postureCache.Upsert(identityId, false, func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData { var postureData *PostureData if exist { - postureData = valueInMap.(*PostureData) + postureData = valueInMap } else { - postureData = newValue.(*PostureData) + postureData = newValue } if _, ok := postureData.ApiSessions[apiSessionId]; !ok { @@ -271,7 +271,6 @@ func shouldPostureCheckTimeoutBeAltered(mfaCheck *persistence.PostureCheckMfa, t return timeoutShouldBeAltered } - return false } diff --git a/controller/model/posture_response_model.go b/controller/model/posture_response_model.go index 0d36a07d4..018078d6d 100644 --- a/controller/model/posture_response_model.go +++ b/controller/model/posture_response_model.go @@ -22,10 +22,10 @@ import ( "github.com/kataras/go-events" "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/controller/persistence" + "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/storage/ast" "github.com/openziti/storage/boltz" - "github.com/openziti/foundation/util/concurrenz" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "go.etcd.io/bbolt" "regexp" "strings" @@ -37,8 +37,8 @@ const ( ) type PostureCache struct { - identityToPostureData cmap.ConcurrentMap //identityId -> PostureData - apiSessionIdToIdentityId cmap.ConcurrentMap //apiSessionId -> identityId + identityToPostureData cmap.ConcurrentMap[*PostureData] + apiSessionIdToIdentityId cmap.ConcurrentMap[string] ticker *time.Ticker isRunning concurrenz.AtomicBoolean events.EventEmmiter @@ -47,8 +47,8 @@ type PostureCache struct { func newPostureCache(env Env) *PostureCache { pc := &PostureCache{ - identityToPostureData: cmap.New(), - apiSessionIdToIdentityId: cmap.New(), + identityToPostureData: cmap.New[*PostureData](), + apiSessionIdToIdentityId: cmap.New[string](), ticker: time.NewTicker(5 * time.Second), EventEmmiter: events.New(), env: env, @@ -162,7 +162,7 @@ func (pc *PostureCache) evaluate() { toDeleteSessionIds = []string{} //notify endpoints that they may have new service posture queries - for identityId, _ := range newIdentityServiceUpdates { + for identityId := range newIdentityServiceUpdates { if _, ok := completedIdentityServiceUpdates[identityId]; !ok { completedIdentityServiceUpdates[identityId] = struct{}{} pc.env.HandleServiceUpdatedEventForIdentityId(identityId) @@ -174,12 +174,13 @@ func (pc *PostureCache) evaluate() { } func (pc *PostureCache) Add(identityId string, postureResponses []*PostureResponse) { - pc.Upsert(identityId, true, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + pc.Upsert(identityId, true, func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData { var postureData *PostureData + if exist { - postureData = valueInMap.(*PostureData) + postureData = valueInMap } else { - postureData = newValue.(*PostureData) + postureData = newValue } for _, postureResponse := range postureResponses { @@ -195,7 +196,7 @@ func (pc *PostureCache) Add(identityId string, postureResponses []*PostureRespon // Upsert is a convenience function to alter the existing PostureData for an identity. If // emitDataAltered is true, posture data listeners will be alerted: this will trigger // service update notifications and posture check evaluation. -func (pc *PostureCache) Upsert(identityId string, emitDataAltered bool, cb func(exist bool, valueInMap interface{}, newValue interface{}) interface{}) { +func (pc *PostureCache) Upsert(identityId string, emitDataAltered bool, cb func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData) { pc.identityToPostureData.Upsert(identityId, newPostureData(), cb) if emitDataAltered { @@ -206,12 +207,13 @@ func (pc *PostureCache) Upsert(identityId string, emitDataAltered bool, cb func( const MaxPostureFailures = 100 func (pc *PostureCache) AddSessionRequestFailure(identityId string, failure *PostureSessionRequestFailure) { - pc.identityToPostureData.Upsert(identityId, newPostureData(), func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + pc.identityToPostureData.Upsert(identityId, newPostureData(), func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData { var postureData *PostureData + if exist { - postureData = valueInMap.(*PostureData) + postureData = valueInMap } else { - postureData = newValue.(*PostureData) + postureData = newValue } postureData.SessionRequestFailures = append(postureData.SessionRequestFailures, failure) @@ -225,8 +227,7 @@ func (pc *PostureCache) AddSessionRequestFailure(identityId string, failure *Pos } func (pc *PostureCache) Evaluate(identityId, apiSessionId string, postureChecks []*PostureCheck) (bool, []*PostureCheckFailure) { - if val, found := pc.identityToPostureData.Get(identityId); found { - postureData := val.(*PostureData) + if postureData, found := pc.identityToPostureData.Get(identityId); found { return postureData.Evaluate(apiSessionId, postureChecks) } @@ -249,12 +250,12 @@ func (pc *PostureCache) Evaluate(identityId, apiSessionId string, postureChecks func (pc *PostureCache) PostureData(identityId string) *PostureData { var result *PostureData = nil - pc.Upsert(identityId, false, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + pc.Upsert(identityId, false, func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData { var pd *PostureData if exist { - pd = valueInMap.(*PostureData) + pd = valueInMap } else { - pd = newValue.(*PostureData) + pd = newValue } result = pd.Copy() @@ -295,15 +296,13 @@ func (pc *PostureCache) ApiSessionDeleted(args ...interface{}) { return } - pc.identityToPostureData.Upsert(apiSession.IdentityId, newPostureData(), func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + pc.identityToPostureData.Upsert(apiSession.IdentityId, newPostureData(), func(exist bool, valueInMap *PostureData, newValue *PostureData) *PostureData { if exist { - pd := valueInMap.(*PostureData) - - if pd != nil && pd.ApiSessions != nil { - delete(pd.ApiSessions, apiSession.Id) + if valueInMap != nil && valueInMap.ApiSessions != nil { + delete(valueInMap.ApiSessions, apiSession.Id) } - return pd + return valueInMap } return newValue @@ -377,7 +376,7 @@ func (pc *PostureCache) PostureCheckChanged(args ...interface{}) { return nil }) - for identityId, _ := range identitiesToNotify { + for identityId := range identitiesToNotify { pc.env.HandleServiceUpdatedEventForIdentityId(identityId) } } @@ -458,7 +457,7 @@ func (pd *PostureData) Evaluate(apiSessionId string, checks []*PostureCheck) (bo func (pd *PostureData) Copy() *PostureData { dest := &PostureData{} - copier.Copy(dest, pd) + _ = copier.Copy(dest, pd) return dest } @@ -503,7 +502,7 @@ type PostureResponseSubType interface { Apply(postureData *PostureData) } -var macClean = regexp.MustCompile("[^a-f0-9]+") +var macClean = regexp.MustCompile("[^a-f\\d]+") func CleanHexString(hexString string) string { return macClean.ReplaceAllString(strings.ToLower(hexString), "") diff --git a/controller/persistence/eventual_eventer.go b/controller/persistence/eventual_eventer.go index c6246ba9e..b9eaafb95 100644 --- a/controller/persistence/eventual_eventer.go +++ b/controller/persistence/eventual_eventer.go @@ -21,9 +21,9 @@ import ( "github.com/kataras/go-events" "github.com/lucsky/cuid" "github.com/michaelquigley/pfxlog" - "github.com/openziti/storage/boltz" "github.com/openziti/foundation/util/concurrenz" - cmap "github.com/orcaman/concurrent-map" + "github.com/openziti/storage/boltz" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/pkg/errors" "go.etcd.io/bbolt" "sync" @@ -277,7 +277,7 @@ const ( // data - []byte - a string array of arguments type EventualEventerBbolt struct { events.EventEmmiter - handlerMap cmap.ConcurrentMap //eventName -> handler func(name, []byte) + handlerMap cmap.ConcurrentMap[[]EventListenerFunc] //eventName -> handlers Interval time.Duration closeNotify <-chan struct{} stopNotify chan struct{} @@ -305,7 +305,7 @@ func NewEventualEventerBbolt(dbProvider DbProvider, store EventualEventStore, in store: store, batchSize: batchSize, trigger: make(chan struct{}, 1), - handlerMap: cmap.New(), + handlerMap: cmap.New[[]EventListenerFunc](), outstandingEvents: &outstanding, } } @@ -354,12 +354,7 @@ func (a *EventualEventerBbolt) AddEventualEvent(eventType string, data []byte) { } func (a *EventualEventerBbolt) AddEventualListener(eventType string, listener EventListenerFunc) { - a.handlerMap.Upsert(eventType, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { - var handlers []EventListenerFunc - if exist { - handlers = valueInMap.([]EventListenerFunc) - } - + a.handlerMap.Upsert(eventType, nil, func(exist bool, handlers []EventListenerFunc, _ []EventListenerFunc) []EventListenerFunc { handlers = append(handlers, listener) return handlers }) @@ -605,8 +600,7 @@ func (a *EventualEventerBbolt) processBatch(info *runInfo, eventualEvents []*Eve info.batchEventIndex++ info.totalEventIndex++ - if handlerVals, found := a.handlerMap.Get(eventualEvent.Type); found { - handlers := handlerVals.([]EventListenerFunc) + if handlers, found := a.handlerMap.Get(eventualEvent.Type); found { for _, handler := range handlers { a.executeHandler(info, eventualEvent, handler) } diff --git a/go.mod b/go.mod index f9c197dd8..0a0fc9286 100644 --- a/go.mod +++ b/go.mod @@ -39,7 +39,7 @@ require ( github.com/openziti/transport/v2 v2.0.3 github.com/openziti/x509-claims v1.0.2 github.com/openziti/xweb v1.0.0 - github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 + github.com/orcaman/concurrent-map/v2 v2.0.0 github.com/pkg/errors v0.9.1 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 github.com/sirupsen/logrus v1.8.1 @@ -94,7 +94,7 @@ require ( github.com/natefinch/lumberjack v2.0.0+incompatible // indirect github.com/oklog/ulid v1.3.1 // indirect github.com/opentracing/opentracing-go v1.2.0 // indirect - github.com/orcaman/concurrent-map/v2 v2.0.0 // indirect + github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 // indirect github.com/parallaxsecond/parsec-client-go v0.0.0-20220111122524-cb78842db373 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect diff --git a/router/fabric/manager.go b/router/fabric/manager.go index 73c667219..8a4b386a1 100644 --- a/router/fabric/manager.go +++ b/router/fabric/manager.go @@ -19,14 +19,14 @@ package fabric import ( "bufio" "fmt" - "google.golang.org/protobuf/proto" "github.com/kataras/go-events" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/edge/runner" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/sirupsen/logrus" + "google.golang.org/protobuf/proto" "math/rand" "sync" "time" @@ -72,10 +72,10 @@ type StateManager interface { } type StateManagerImpl struct { - apiSessionsByToken *sync.Map // apiSesion token -> *edge_ctrl_pb.ApiSession - activeApiSessions cmap.ConcurrentMap // apiSession token -> MapWithMutex[session token] -> func(){} - sessions cmap.ConcurrentMap // session token -> uint32 - recentlyRemovedSessions cmap.ConcurrentMap // session token -> time.Time (time added, time.Now()) + apiSessionsByToken cmap.ConcurrentMap[*edge_ctrl_pb.ApiSession] + activeApiSessions cmap.ConcurrentMap[*MapWithMutex] + sessions cmap.ConcurrentMap[uint32] + recentlyRemovedSessions cmap.ConcurrentMap[time.Time] Hostname string ControllerAddr string @@ -111,10 +111,10 @@ func (sm *StateManagerImpl) IsSyncInProgress() bool { func NewStateManager() StateManager { return &StateManagerImpl{ EventEmmiter: events.New(), - apiSessionsByToken: &sync.Map{}, - activeApiSessions: cmap.New(), - sessions: cmap.New(), - recentlyRemovedSessions: cmap.New(), + apiSessionsByToken: cmap.New[*edge_ctrl_pb.ApiSession](), + activeApiSessions: cmap.New[*MapWithMutex](), + sessions: cmap.New[uint32](), + recentlyRemovedSessions: cmap.New[time.Time](), } } @@ -124,7 +124,7 @@ func (sm *StateManagerImpl) AddApiSession(apiSession *edge_ctrl_pb.ApiSession) { WithField("apiSessionToken", apiSession.Token). WithField("apiSessionCertFingerprints", apiSession.CertFingerprints). Debugf("adding apiSession [id: %s] [token: %s] fingerprints [%s]", apiSession.Id, apiSession.Token, apiSession.CertFingerprints) - sm.apiSessionsByToken.Store(apiSession.Token, apiSession) + sm.apiSessionsByToken.Set(apiSession.Token, apiSession) sm.Emit(EventAddedApiSession, apiSession) } @@ -134,14 +134,14 @@ func (sm *StateManagerImpl) UpdateApiSession(apiSession *edge_ctrl_pb.ApiSession WithField("apiSessionToken", apiSession.Token). WithField("apiSessionCertFingerprints", apiSession.CertFingerprints). Debugf("updating apiSession [id: %s] [token: %s] fingerprints [%s]", apiSession.Id, apiSession.Token, apiSession.CertFingerprints) - sm.apiSessionsByToken.Store(apiSession.Token, apiSession) + sm.apiSessionsByToken.Set(apiSession.Token, apiSession) sm.Emit(EventUpdatedApiSession, apiSession) } func (sm *StateManagerImpl) RemoveApiSession(token string) { - if ns, ok := sm.apiSessionsByToken.Load(token); ok { + if ns, ok := sm.apiSessionsByToken.Get(token); ok { pfxlog.Logger().WithField("apiSessionToken", token).Debugf("removing api session [token: %s]", token) - sm.apiSessionsByToken.Delete(token) + sm.apiSessionsByToken.Remove(token) eventName := sm.getApiSessionRemovedEventName(token) sm.Emit(eventName) sm.RemoveAllListeners(eventName) @@ -151,9 +151,9 @@ func (sm *StateManagerImpl) RemoveApiSession(token string) { } } -// Removes API Sessions not present in the knownApiSessions argument. If the beforeSessionId value is not empty string, -// it will be used as a monotonic comparison between it and API session ids. API session ids later than the sync -// will be ignored. +// RemoveMissingApiSessions removes API Sessions not present in the knownApiSessions argument. If the beforeSessionId +// value is not empty string, it will be used as a monotonic comparison between it and API session ids. API session ids +// later than the sync will be ignored. func (sm *StateManagerImpl) RemoveMissingApiSessions(knownApiSessions []*edge_ctrl_pb.ApiSession, beforeSessionId string) { validTokens := map[string]bool{} for _, apiSession := range knownApiSessions { @@ -161,14 +161,10 @@ func (sm *StateManagerImpl) RemoveMissingApiSessions(knownApiSessions []*edge_ct } var tokensToRemove []string - sm.apiSessionsByToken.Range(func(key, val interface{}) bool { - token, _ := key.(string) - apiSession, _ := val.(*edge_ctrl_pb.ApiSession) - + sm.apiSessionsByToken.IterCb(func(token string, apiSession *edge_ctrl_pb.ApiSession) { if _, ok := validTokens[token]; !ok && (beforeSessionId == "" || apiSession.Id <= beforeSessionId) { tokensToRemove = append(tokensToRemove, token) } - return true }) for _, token := range tokensToRemove { @@ -182,7 +178,7 @@ func (sm *StateManagerImpl) RemoveEdgeSession(token string) { sm.Emit(eventName) sm.RemoveAllListeners(eventName) - sm.sessions.RemoveCb(token, func(key string, v interface{}, exists bool) bool { + sm.sessions.RemoveCb(token, func(key string, _ uint32, exists bool) bool { if exists { sm.recentlyRemovedSessions.Set(token, time.Now()) } @@ -213,10 +209,8 @@ func (sm *StateManagerImpl) GetApiSessionWithTimeout(token string, timeout time. } func (sm *StateManagerImpl) GetApiSession(token string) *edge_ctrl_pb.ApiSession { - if val, ok := sm.apiSessionsByToken.Load(token); ok { - if session, ok := val.(*edge_ctrl_pb.ApiSession); ok { - return session - } + if apiSession, ok := sm.apiSessionsByToken.Get(token); ok { + return apiSession } return nil } @@ -227,11 +221,11 @@ func (sm *StateManagerImpl) AddEdgeSessionRemovedListener(token string, callBack return func() {} } - sm.sessions.Upsert(token, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + sm.sessions.Upsert(token, 0, func(exist bool, valueInMap uint32, newValue uint32) uint32 { if !exist { return uint32(1) } - return valueInMap.(uint32) + 1 + return valueInMap + 1 }) eventName := sm.getEdgeSessionRemovedEventName(token) @@ -251,19 +245,19 @@ func (sm *StateManagerImpl) AddEdgeSessionRemovedListener(token string, callBack } func (sm *StateManagerImpl) SessionConnectionClosed(token string) { - sm.sessions.Upsert(token, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + sm.sessions.Upsert(token, 0, func(exist bool, valueInMap uint32, newValue uint32) uint32 { if !exist { return uint32(0) } - return valueInMap.(uint32) + 1 + return valueInMap + 1 }) - sm.sessions.RemoveCb(token, func(key string, v interface{}, exists bool) bool { + sm.sessions.RemoveCb(token, func(key string, v uint32, exists bool) bool { if !exists { return false } - if v.(uint32) == 0 { + if v == 0 { sm.recentlyRemovedSessions.Set(token, time.Now()) return true } @@ -318,7 +312,7 @@ func (sm *StateManagerImpl) StartHeartbeat(ctrl channel.Channel, intervalSeconds } func (sm *StateManagerImpl) AddConnectedApiSession(token string) { - sm.activeApiSessions.Upsert(token, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + sm.activeApiSessions.Upsert(token, nil, func(exist bool, valueInMap *MapWithMutex, newValue *MapWithMutex) *MapWithMutex { if exist { return valueInMap } @@ -334,10 +328,7 @@ func (sm *StateManagerImpl) AddConnectedApiSessionWithChannel(token string, remo var sessions *MapWithMutex for sessions == nil { - if val, ok := sm.activeApiSessions.Get(token); ok { - if sessions, ok = val.(*MapWithMutex); !ok { - pfxlog.Logger().Panic("could not convert to active sessions") - } + if _, ok := sm.activeApiSessions.Get(token); ok { sessions.Put(ch, removeCB) } else { sessions = newMapWithMutex() @@ -350,9 +341,7 @@ func (sm *StateManagerImpl) AddConnectedApiSessionWithChannel(token string, remo } func (sm *StateManagerImpl) RemoveConnectedApiSessionWithChannel(token string, ch channel.Channel) { - if val, ok := sm.activeApiSessions.Get(token); ok { - sessions, ok := val.(*MapWithMutex) - + if sessions, ok := sm.activeApiSessions.Get(token); ok { if !ok { pfxlog.Logger().Panic("could not convert active sessions to map") } @@ -378,13 +367,10 @@ func (sm *StateManagerImpl) ActiveApiSessionTokens() []string { func (sm *StateManagerImpl) flushRecentlyRemoved() { now := time.Now() var toRemove []string - sm.recentlyRemovedSessions.IterCb(func(key string, v interface{}) { + sm.recentlyRemovedSessions.IterCb(func(key string, t time.Time) { remove := false - if t, ok := v.(time.Time); ok { - if now.Sub(t) >= time.Minute { - remove = true - } - } else { + + if now.Sub(t) >= time.Minute { remove = true } @@ -406,22 +392,23 @@ func (sm *StateManagerImpl) DumpApiSessions(c *bufio.ReadWriter) error { i := 0 deadline := time.After(time.Second) timedOut := false - sm.apiSessionsByToken.Range(func(key, value interface{}) bool { + + for _, session := range sm.apiSessionsByToken.Items() { i++ - session := value.(*edge_ctrl_pb.ApiSession) val := fmt.Sprintf("%v: id: %v, token: %v\n", i, session.Id, session.Token) select { case ch <- val: case <-deadline: timedOut = true - return false + break } if i%10000 == 0 { // allow a second to dump each 10k entries deadline = time.After(time.Second) } - return true - }) + + } + if timedOut { select { case ch <- "timed out": diff --git a/router/xgress_edge_tunnel/dialer.go b/router/xgress_edge_tunnel/dialer.go index 88f6e039b..44bc7f06c 100644 --- a/router/xgress_edge_tunnel/dialer.go +++ b/router/xgress_edge_tunnel/dialer.go @@ -41,11 +41,10 @@ func (self *tunneler) Dial(destination string, circuitId *identity.TokenId, addr WithField("binding", "edge"). WithField("destination", destination) - val, ok := self.terminators.Get(destination) + terminator, ok := self.terminators.Get(destination) if !ok { return nil, xgress.InvalidTerminatorError{InnerError: errors.Errorf("tunnel terminator for destination %v not found", destination)} } - terminator := val.(*tunnelTerminator) options, err := tunnel.AppDataToMap(circuitId.Data[edge.AppDataHeader]) if err != nil { diff --git a/router/xgress_edge_tunnel/fabric.go b/router/xgress_edge_tunnel/fabric.go index 9d2284d2b..fde94cce6 100644 --- a/router/xgress_edge_tunnel/fabric.go +++ b/router/xgress_edge_tunnel/fabric.go @@ -33,7 +33,7 @@ import ( "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/sdk-golang/ziti/edge" "github.com/openziti/sdk-golang/ziti/sdkinfo" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/sirupsen/logrus" "net" "runtime" @@ -46,8 +46,8 @@ func newProvider(factory *Factory, tunneler *tunneler) *fabricProvider { return &fabricProvider{ factory: factory, tunneler: tunneler, - dialSessions: cmap.New(), - bindSessions: cmap.New(), + dialSessions: cmap.New[string](), + bindSessions: cmap.New[string](), } } @@ -59,24 +59,18 @@ type fabricProvider struct { apiSessionToken string currentIdentity *edge.CurrentIdentity - dialSessions cmap.ConcurrentMap - bindSessions cmap.ConcurrentMap + dialSessions cmap.ConcurrentMap[string] + bindSessions cmap.ConcurrentMap[string] } func (self *fabricProvider) getDialSession(serviceName string) string { sessionId, _ := self.dialSessions.Get(serviceName) - if sessionId != nil { - return sessionId.(string) - } - return "" + return sessionId } func (self *fabricProvider) getBindSession(serviceName string) string { sessionId, _ := self.bindSessions.Get(serviceName) - if sessionId != nil { - return sessionId.(string) - } - return "" + return sessionId } func (self *fabricProvider) updateApiSession(resp *edge_ctrl_pb.CreateApiSessionResponse) { diff --git a/router/xgress_edge_tunnel/servicepoll.go b/router/xgress_edge_tunnel/servicepoll.go index 3d2cbd882..f37e593f3 100644 --- a/router/xgress_edge_tunnel/servicepoll.go +++ b/router/xgress_edge_tunnel/servicepoll.go @@ -19,7 +19,7 @@ import ( "github.com/openziti/edge/tunnel/intercept" "github.com/openziti/sdk-golang/ziti" "github.com/openziti/sdk-golang/ziti/edge" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/sirupsen/logrus" "reflect" "sync" @@ -29,7 +29,7 @@ import ( func newServicePoller(fabricProvider *fabricProvider) *servicePoller { result := &servicePoller{ - services: cmap.New(), + services: cmap.New[*edge.Service](), fabricProvider: fabricProvider, } @@ -37,7 +37,7 @@ func newServicePoller(fabricProvider *fabricProvider) *servicePoller { } type servicePoller struct { - services cmap.ConcurrentMap + services cmap.ConcurrentMap[*edge.Service] serviceListener *intercept.ServiceListener servicesLastUpdateToken atomic.Value serviceListenerLock sync.Mutex @@ -66,8 +66,7 @@ func (self *servicePoller) handleServiceListUpdate(lastUpdateToken []byte, servi var toRemove []string // process Deletes - self.services.IterCb(func(k string, value interface{}) { - svc := value.(*edge.Service) + self.services.IterCb(func(k string, svc *edge.Service) { if _, found := idMap[svc.Id]; !found { toRemove = append(toRemove, k) self.serviceListener.HandleServicesChange(ziti.ServiceRemoved, svc) @@ -80,7 +79,7 @@ func (self *servicePoller) handleServiceListUpdate(lastUpdateToken []byte, servi // Adds and Updates for _, s := range services { - self.services.Upsert(s.Id, s, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + self.services.Upsert(s.Id, s, func(exist bool, valueInMap *edge.Service, newValue *edge.Service) *edge.Service { if !exist { self.serviceListener.HandleServicesChange(ziti.ServiceAdded, s) return s diff --git a/router/xgress_edge_tunnel/tunneler.go b/router/xgress_edge_tunnel/tunneler.go index bbf673288..643783b14 100644 --- a/router/xgress_edge_tunnel/tunneler.go +++ b/router/xgress_edge_tunnel/tunneler.go @@ -25,7 +25,7 @@ import ( "github.com/openziti/edge/tunnel/intercept/proxy" "github.com/openziti/edge/tunnel/intercept/tproxy" "github.com/openziti/fabric/router/xgress" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/pkg/errors" "net" "time" @@ -40,13 +40,13 @@ type tunneler struct { interceptor intercept.Interceptor servicePoller *servicePoller fabricProvider *fabricProvider - terminators cmap.ConcurrentMap + terminators cmap.ConcurrentMap[*tunnelTerminator] } func newTunneler(factory *Factory, stateManager fabric.StateManager) *tunneler { result := &tunneler{ stateManager: stateManager, - terminators: cmap.New(), + terminators: cmap.New[*tunnelTerminator](), } result.fabricProvider = newProvider(factory, result) @@ -101,17 +101,16 @@ func (self *tunneler) removeStaleConnections(notifyClose <-chan struct{}) { select { case <-ticker.C: var toRemove []string - self.terminators.IterCb(func(key string, v interface{}) { - if t, ok := v.(*tunnelTerminator); ok { - if t.closed.Get() { - toRemove = append(toRemove, key) - } + self.terminators.IterCb(func(key string, t *tunnelTerminator) { + + if t.closed.Get() { + toRemove = append(toRemove, key) } + }) for _, key := range toRemove { - if v, found := self.terminators.Get(key); found { - t := v.(*tunnelTerminator) + if t, found := self.terminators.Get(key); found { self.terminators.Remove(key) pfxlog.Logger().Debugf("removed closed tunnel terminator %v for service %v", key, t.context.ServiceName()) } diff --git a/tunnel/dns/refcount.go b/tunnel/dns/refcount.go index 101d1459a..b801a53f6 100644 --- a/tunnel/dns/refcount.go +++ b/tunnel/dns/refcount.go @@ -1,19 +1,19 @@ package dns import ( - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "net" ) func NewRefCountingResolver(resolver Resolver) Resolver { return &RefCountingResolver{ - names: cmap.New(), + names: cmap.New[int](), wrapped: resolver, } } type RefCountingResolver struct { - names cmap.ConcurrentMap + names cmap.ConcurrentMap[int] wrapped Resolver } @@ -28,9 +28,9 @@ func (self *RefCountingResolver) AddDomain(name string, cb func(string) (net.IP, func (self *RefCountingResolver) AddHostname(s string, ip net.IP) error { err := self.wrapped.AddHostname(s, ip) if err != nil { - self.names.Upsert(s, 1, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + self.names.Upsert(s, 1, func(exist bool, valueInMap int, newValue int) int { if exist { - return valueInMap.(int) + 1 + return valueInMap + 1 } return 1 }) @@ -39,14 +39,14 @@ func (self *RefCountingResolver) AddHostname(s string, ip net.IP) error { } func (self *RefCountingResolver) RemoveHostname(s string) error { - val := self.names.Upsert(s, 1, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + val := self.names.Upsert(s, 1, func(exist bool, valueInMap int, newValue int) int { if exist { - return valueInMap.(int) - 1 + return valueInMap - 1 } return 0 }) - if result := val.(int); result == 0 { + if val == 0 { self.names.Remove(s) return self.wrapped.RemoveHostname(s) } diff --git a/tunnel/intercept/tproxy/tproxy_linux.go b/tunnel/intercept/tproxy/tproxy_linux.go index 1892eb047..8604b44dc 100644 --- a/tunnel/intercept/tproxy/tproxy_linux.go +++ b/tunnel/intercept/tproxy/tproxy_linux.go @@ -31,7 +31,7 @@ import ( "github.com/openziti/foundation/util/mempool" "github.com/openziti/foundation/util/stringz" "github.com/openziti/sdk-golang/ziti/edge/impl" - cmap "github.com/orcaman/concurrent-map" + cmap "github.com/orcaman/concurrent-map/v2" "github.com/pkg/errors" "github.com/sirupsen/logrus" "golang.org/x/sys/unix" @@ -76,7 +76,7 @@ func New(lanIf string) (intercept.Interceptor, error) { return &interceptor{ lanIf: lanIf, - serviceProxies: cmap.New(), + serviceProxies: cmap.New[*tProxy](), ipt: ipt, }, nil } @@ -92,7 +92,7 @@ func (a alwaysRemoveAddressTracker) RemoveAddress(string) bool { type interceptor struct { lanIf string provider tunnel.FabricProvider - serviceProxies cmap.ConcurrentMap + serviceProxies cmap.ConcurrentMap[*tProxy] ipt *iptables.IPTables } @@ -102,8 +102,7 @@ func (self *interceptor) Start(provider tunnel.FabricProvider) { func (self *interceptor) Stop() { servicesRemoved := false - self.serviceProxies.IterCb(func(key string, v interface{}) { - proxy := v.(*tProxy) + self.serviceProxies.IterCb(func(key string, proxy *tProxy) { proxy.Stop(alwaysRemoveAddressTracker{}) servicesRemoved = true }) @@ -124,8 +123,7 @@ func (self *interceptor) Intercept(service *entities.Service, resolver dns.Resol } func (self *interceptor) StopIntercepting(serviceName string, tracker intercept.AddressTracker) error { - if val, found := self.serviceProxies.Get(serviceName); found { - proxy := val.(*tProxy) + if proxy, found := self.serviceProxies.Get(serviceName); found { proxy.Stop(tracker) self.serviceProxies.Remove(serviceName) self.cleanupChains()