Merge pull request #1843 from openziti/ha-smoke

Update HA smoketest for HA changes

controller/raft/mesh/mesh.go

@@ -421,8 +421,8 @@ func (self *impl) GetPeer(addr raft.ServerAddress) *Peer {
 }
 
 func (self *impl) PeerDisconnected(peer *Peer) {
-	self.lock.RLock()
-	defer self.lock.RUnlock()
+	self.lock.Lock()
+	defer self.lock.Unlock()
 	currentPeer := self.Peers[peer.Address]
 	if currentPeer == nil || currentPeer != peer {
 		return

etc/ctrl.with.edge.yml

@@ -243,6 +243,7 @@ web:
         options: { }
       - binding: edge-oidc
         options:
+          secret: 38de0de7fa283e8c8ef2a7823a6d3e727681ade50b9eb7bee2424197fb22bf18
           redirectURIs:
             - "http://localhost:*/auth/callback"
             - "http://127.0.0.1:*/auth/callback"

router/state/manager.go

@@ -441,9 +441,10 @@ func (sm *ManagerImpl) GetApiSession(token string) *ApiSession {
 				JwtToken: jwtToken,
 				Claims:   accessClaims,
 			}
+		} else {
+			pfxlog.Logger().WithError(err).Error("JWT validation failed")
+			return nil
 		}
-
-		//fall through to check if the token is a zt-session
 	}
 
 	if apiSession, ok := sm.apiSessionsByToken.Get(token); ok {

router/xgress_edge/factory.go

@@ -137,7 +137,7 @@ func (factory *Factory) LoadConfig(configMap map[interface{}]interface{}) error
 func NewFactory(routerConfig *router.Config, env env.RouterEnv, stateManager state.Manager) *Factory {
 	factory := &Factory{
 		ctrls:           env.GetNetworkControllers(),
-		hostedServices:  newHostedServicesRegistry(env),
+		hostedServices:  newHostedServicesRegistry(env, stateManager),
 		stateManager:    stateManager,
 		versionProvider: env.GetVersionInfo(),
 		routerConfig:    routerConfig,

router/xgress_edge/hosted.go

@@ -28,19 +28,22 @@ import (
 	"github.com/openziti/ziti/controller/apierror"
 	"github.com/openziti/ziti/controller/command"
 	routerEnv "github.com/openziti/ziti/router/env"
+	"github.com/openziti/ziti/router/state"
 	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/proto"
+	"strings"
 	"sync/atomic"
 	"time"
 )
 
-func newHostedServicesRegistry(env routerEnv.RouterEnv) *hostedServiceRegistry {
+func newHostedServicesRegistry(env routerEnv.RouterEnv, stateManager state.Manager) *hostedServiceRegistry {
 	result := &hostedServiceRegistry{
 		terminators:  cmap.New[*edgeTerminator](),
 		events:       make(chan terminatorEvent),
 		env:          env,
+		stateManager: stateManager,
 		triggerEvalC: make(chan struct{}, 1),
 		establishSet: map[string]*edgeTerminator{},
 		deleteSet:    map[string]*edgeTerminator{},
@@ -53,6 +56,7 @@ type hostedServiceRegistry struct {
 	terminators  cmap.ConcurrentMap[string, *edgeTerminator]
 	events       chan terminatorEvent
 	env          routerEnv.RouterEnv
+	stateManager state.Manager
 	establishSet map[string]*edgeTerminator
 	deleteSet    map[string]*edgeTerminator
 	triggerEvalC chan struct{}
@@ -546,6 +550,16 @@ func (self *hostedServiceRegistry) establishTerminator(terminator *edgeTerminato
 		InstanceSecret: terminator.instanceSecret,
 	}
 
+	if self.stateManager.GetConfig().Ha.Enabled && strings.HasPrefix(request.SessionToken, JwtTokenPrefix) {
+		apiSession := self.stateManager.GetApiSessionFromCh(terminator.Channel)
+
+		if apiSession == nil {
+			return errors.New("could not find api session for channel, unable to process bind message")
+		}
+
+		request.ApiSessionToken = apiSession.Token
+	}
+
 	ctrlCh := factory.ctrls.AnyCtrlChannel()
 	if ctrlCh == nil {
 		errStr := "no controller available, cannot create terminator"

router/xgress_edge_tunnel/fabric.go

@@ -19,7 +19,6 @@ package xgress_edge_tunnel
 import (
 	"encoding/json"
 	"github.com/cenkalti/backoff/v4"
-	"github.com/google/uuid"
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/channel/v2"
 	"github.com/openziti/channel/v2/protobufs"
@@ -33,6 +32,7 @@ import (
 	"github.com/openziti/ziti/common/build"
 	"github.com/openziti/ziti/common/ctrl_msg"
 	"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
+	"github.com/openziti/ziti/controller/idgen"
 	"github.com/openziti/ziti/router/xgress"
 	"github.com/openziti/ziti/router/xgress_common"
 	"github.com/openziti/ziti/tunnel"
@@ -289,7 +289,7 @@ func (self *fabricProvider) TunnelService(service tunnel.Service, terminatorInst
 }
 
 func (self *fabricProvider) HostService(hostCtx tunnel.HostingContext) (tunnel.HostControl, error) {
-	id := uuid.NewString()
+	id := idgen.NewUUIDString()
 
 	terminator := &tunnelTerminator{
 		id:            id,

ziti/cmd/demo/echo_server.go

@@ -60,6 +60,7 @@ type echoServer struct {
 	cliAgentEnabled bool
 	cliAgentAddr    string
 	cliAgentAlias   string
+	ha              bool
 }
 
 func newEchoServerCmd() *cobra.Command {
@@ -84,6 +85,7 @@ func newEchoServerCmd() *cobra.Command {
 	cmd.Flags().BoolVar(&server.cliAgentEnabled, "cli-agent", true, "Enable/disable CLI Agent (enabled by default)")
 	cmd.Flags().StringVar(&server.cliAgentAddr, "cli-agent-addr", "", "Specify where CLI Agent should list (ex: unix:/tmp/myfile.sock or tcp:127.0.0.1:10001)")
 	cmd.Flags().StringVar(&server.cliAgentAlias, "cli-agent-alias", "", "Alias which can be used by ziti agent commands to find this instance")
+	cmd.Flags().BoolVar(&server.ha, "ha", false, "Enable HA controller compatibility")
 
 	return cmd
 }
@@ -157,6 +159,7 @@ func (self *echoServer) run(*cobra.Command, []string) {
 		if err != nil {
 			log.WithError(err).Fatalf("ziti: unable to load ziti identity from [%v]", self.configFile)
 		}
+		zitiConfig.EnableHa = self.ha
 		self.zitiIdentity, err = identity.LoadIdentity(zitiConfig.ID)
 		if err != nil {
 			log.WithError(err).Fatalf("ziti: unable to create ziti identity from [%v]", self.configFile)
@@ -168,6 +171,10 @@ func (self *echoServer) run(*cobra.Command, []string) {
 			log.WithError(err).Fatal("unable to get create ziti context from config")
 		}
 
+		if self.ha {
+			zitiContext.(*ziti.ContextImpl).CtrlClt.SetUseOidc(true)
+		}
+
 		zitiIdentity, err := zitiContext.GetCurrentIdentity()
 		if err != nil {
 			log.WithError(err).Fatal("unable to get current ziti identity from controller")

ziti/cmd/demo/zcat.go

@@ -34,6 +34,7 @@ type zcatAction struct {
 	verbose      bool
 	logFormatter string
 	configFile   string
+	ha           bool // todo: remove when ha flag is no longer needed
 }
 
 func newZcatCmd() *cobra.Command {
@@ -52,6 +53,7 @@ func newZcatCmd() *cobra.Command {
 	cmd.Flags().BoolVarP(&action.verbose, "verbose", "v", false, "Enable verbose logging")
 	cmd.Flags().StringVar(&action.logFormatter, "log-formatter", "", "Specify log formatter [json|pfxlog|text]")
 	cmd.Flags().StringVarP(&action.configFile, "identity", "i", "", "Specify the Ziti identity to use. If not specified the Ziti listener won't be started")
+	cmd.Flags().BoolVar(&action.ha, "ha", false, "Enable HA controller compatibility")
 	cmd.Flags().SetInterspersed(true)
 
 	return cmd
@@ -107,6 +109,7 @@ func (self *zcatAction) run(_ *cobra.Command, args []string) {
 			dialIdentifier = addr[:atIdx]
 			addr = addr[atIdx+1:]
 		}
+		zitiConfig.EnableHa = self.ha
 
 		zitiContext, ctxErr := ziti.NewContext(zitiConfig)
 		if ctxErr != nil {

zititest/models/simple/configs/ctrl.yml.tmpl

@@ -188,13 +188,7 @@ web:
       #   - edge-client
       #   - fabric-management
       - binding: health-checks
-        options: {}
       - binding: fabric
       - binding: edge-management
-        # options - variable optional/required
-        # This section is used to define values that are specified by the API they are associated with.
-        # These settings are per API. The example below is for the `edge-api` and contains both optional values and
-        # required values.
-        options: {}
       - binding: edge-client
-        options: {}
+      - binding: edge-oidc

zititest/models/simple/configs/router.yml.tmpl

@@ -6,6 +6,11 @@ v: 3
 
 enableDebugOps: true
 
+{{if .Component.GetFlag "ha"}}
+ha:
+  enabled: true
+{{end}}
+
 identity:
   cert:                 /home/{{$ssh_username}}/fablab/cfg/{{$identity}}-client.cert
   server_cert:          /home/{{$ssh_username}}/fablab/cfg/{{$identity}}-server.cert

zititest/models/simple/simple.go

@@ -38,7 +38,7 @@ import (
 	"time"
 )
 
-const ZitiEdgeTunnelVersion = "v0.22.17"
+const ZitiEdgeTunnelVersion = "v0.22.25"
 
 //go:embed configs
 var configResource embed.FS

zititest/tests/echo_test.go

@@ -38,9 +38,13 @@ func TestSdkEcho(t *testing.T) {
 	for _, c := range components {
 		remoteConfigFile := "/home/ubuntu/fablab/cfg/" + c.Id + ".json"
 
-		echoClientCmd := fmt.Sprintf(`echo "%s" | /home/%s/fablab/bin/ziti demo zcat --identity %s ziti:echo 2>&1`,
-			string(data), c.GetHost().GetSshUser(), remoteConfigFile)
-
+		ha := ""
+		if len(run.GetModel().SelectComponents(".ctrl")) > 1 {
+			ha = "--ha"
+		}
+		echoClientCmd := fmt.Sprintf(`echo "%s" | /home/%s/fablab/bin/ziti demo zcat %s --identity %s ziti:echo 2>&1`,
+			data, c.GetHost().GetSshUser(), ha, remoteConfigFile)
+		t.Logf("running: %s", echoClientCmd)
 		output, err := c.GetHost().ExecLogged(echoClientCmd)
 		t.Logf("test output:\n%s", output)
 		req.NoError(err)

zititest/tests/scp_test.go

@@ -85,7 +85,7 @@ func testScp(t *testing.T, hostSelector string, hostType string, encrypted bool)
 		encDesk = "unencrypted"
 	}
 
-	success := false
+	success := true
 
 	nameExtra := ""
 	if !encrypted {
@@ -107,6 +107,7 @@ func testScp(t *testing.T, hostSelector string, hostType string, encrypted bool)
 
 	for _, test := range tests {
 		t.Run(fmt.Sprintf("(%s%s%s)-%v", hostSelector, test.direction, hostType, encDesk), func(t *testing.T) {
+			success = false
 			host, err := model.GetModel().SelectHost("." + hostSelector + "-client")
 			req := require.New(t)
 			req.NoError(err)

zititest/zitilab/component_echo_server.go

@@ -45,15 +45,20 @@ func (self *EchoServerType) IsRunning(_ model.Run, c *model.Component) (bool, er
 	return len(pids) > 0, nil
 }
 
-func (self *EchoServerType) Start(_ model.Run, c *model.Component) error {
+func (self *EchoServerType) Start(run model.Run, c *model.Component) error {
 	user := c.GetHost().GetSshUser()
 
 	binaryPath := getZitiBinaryPath(c, self.Version)
 	configPath := fmt.Sprintf("/home/%s/fablab/cfg/%s.json", user, c.Id)
 	logsPath := fmt.Sprintf("/home/%s/logs/%s.log", user, c.Id)
 
-	serviceCmd := fmt.Sprintf("nohup %s demo echo-server -i %s --cli-agent-alias %s > %s 2>&1 &",
-		binaryPath, configPath, c.Id, logsPath)
+	ha := ""
+	if len(run.GetModel().SelectComponents(".ctrl")) > 1 {
+		ha = "--ha"
+	}
+
+	serviceCmd := fmt.Sprintf("nohup %s demo echo-server --cli-agent-alias %s %s -i %s > %s 2>&1 &",
+		binaryPath, c.Id, ha, configPath, logsPath)
 
 	value, err := c.GetHost().ExecLogged(serviceCmd)
 	if err != nil {