gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-13 07:41:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1023 from openziti/use.cmap.v2

Browse Source 
use cmap generics, update fabric
This commit is contained in:
Andrew
2022-05-27 14:12:20 -04:00
committed by GitHub
parent 1af4e6cdf6 e74c76c44b
commit fb1214b65f
20 changed files with 172 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/env/appenv.go
Vendored
+3 -3
View File
@@ -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(),
}
controller/env/trace.go
Vendored
+7 -7
View File
@@ -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)
}
controller/handler_edge_ctrl/list_tunnel_services.go
+2 -2
View File
@@ -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
}
controller/internal/routes/current_api_session_router.go
+1 -1
View File
@@ -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
}
controller/model/api_session_heartbeats.go
+14 -18
View File
@@ -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,
})
}
})
controller/model/authenticator_mod_cert.go
+8 -10
View File
@@ -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)
controller/model/authenticator_mod_ext_jwt.go
+4 -12
View File
@@ -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)
controller/model/authenticator_mod_updb.go
+5 -6
View File
@@ -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 {
controller/model/identity_handlers.go
+13 -15
View File
@@ -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)
}
}
}
controller/model/posture_response_handlers.go
+6 -7
View File
@@ -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
}
controller/model/posture_response_model.go
+27 -28
View File
@@ -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), "")
controller/persistence/eventual_eventer.go
+6 -12
View File
@@ -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)
}
go.mod
+2 -2
View File
@@ -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
router/fabric/manager.go
+39 -52
View File
@@ -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":
router/xgress_edge_tunnel/dialer.go
+1 -2
View File
@@ -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 {
router/xgress_edge_tunnel/fabric.go
+7 -13
View File
@@ -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) {
router/xgress_edge_tunnel/servicepoll.go
+5 -6
View File
@@ -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
router/xgress_edge_tunnel/tunneler.go
+9 -10
View File
@@ -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())
}
tunnel/dns/refcount.go
+8 -8
View File
@@ -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)
}
tunnel/intercept/tproxy/tproxy_linux.go
+5 -7
View File
@@ -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()