Merge branch 'release/v1.1' into feature/new-config-options

flomesh-io · Oct 27, 2023 · 67397f8 · 67397f8
2 parents a9258c8 + fed84de
commit 67397f8
Show file tree

Hide file tree

Showing 11 changed files with 712 additions and 62 deletions.
diff --git a/charts/fsm/components/scripts/gateways/extension/agent-service.js b/charts/fsm/components/scripts/gateways/extension/agent-service.js
@@ -0,0 +1,15 @@
+((
+) => pipy()
+
+.pipeline()
+.handleMessageStart(
+  msg => (
+    msg?.head?.headers?.['hy-agent-rsn'] && (
+      msg.head.headers['orig-host'] = msg.head.headers.host,
+      msg.head.headers.host = msg.head.headers['hy-agent-rsn']
+    )
+  )
+)
+.chain()
+
+)()
diff --git a/charts/fsm/templates/fsm-consul-connector-deployment.yaml b/charts/fsm/templates/fsm-consul-connector-deployment.yaml
@@ -57,6 +57,7 @@ spec:
             "-mesh-name", "{{.Values.fsm.meshName}}",
             "-trust-domain", "{{.Values.fsm.trustDomain}}",
             "-derive-namespace={{.Values.fsm.cloudConnector.consul.deriveNamespace}}",
+            "-with-gateway-api={{.Values.fsm.fsmGateway.enabled}}",
             "-http-addr={{.Values.fsm.cloudConnector.consul.httpAddr}}",
             "-passing-only={{.Values.fsm.cloudConnector.consul.passingOnly}}",
             "-filter-tag={{.Values.fsm.cloudConnector.consul.filterTag}}",

diff --git a/charts/fsm/templates/fsm-eureka-connector-deployment.yaml b/charts/fsm/templates/fsm-eureka-connector-deployment.yaml
@@ -57,6 +57,7 @@ spec:
             "-mesh-name", "{{.Values.fsm.meshName}}",
             "-trust-domain", "{{.Values.fsm.trustDomain}}",
             "-derive-namespace={{.Values.fsm.cloudConnector.eureka.deriveNamespace}}",
+            "-with-gateway-api={{.Values.fsm.fsmGateway.enabled}}",
             "-http-addr={{.Values.fsm.cloudConnector.eureka.httpAddr}}",
             "-passing-only={{.Values.fsm.cloudConnector.eureka.passingOnly}}",
             "-filter-tag={{.Values.fsm.cloudConnector.eureka.filterTag}}",

diff --git a/charts/fsm/templates/fsm-rbac.yaml b/charts/fsm/templates/fsm-rbac.yaml
@@ -29,10 +29,10 @@ rules:
   - apiGroups: [""]
     resources: ["nodes"]
     verbs: ["get", "list", "watch"]
-{{- if semverCompare ">=1.21-0" .Capabilities.KubeVersion.GitVersion }}
+{{- if (semverCompare ">=1.21-0" .Capabilities.KubeVersion.GitVersion) }}
   - apiGroups: ["discovery.k8s.io"]
     resources: ["endpointslices"]
-    verbs: ["get", "list", "watch", "update", "patch"]
+    verbs: ["get", "list", "watch", "create", "update", "patch"]
 {{- end }}
 
   # Port forwarding is needed for the FSM pod to be able to connect

diff --git a/cmd/fsm-connector/consul/fsm-consul-connector.go b/cmd/fsm-connector/consul/fsm-consul-connector.go
@@ -12,6 +12,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	gwapi "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned"
 
 	"github.com/hashicorp/consul/command/flags"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -48,6 +49,7 @@ var (
 	prefixTag         string
 	suffixTag         string
 	deriveNamespace   string
+	withGatewayAPI    bool
 
 	scheme = runtime.NewScheme()
 )
@@ -72,6 +74,7 @@ func init() {
 	cliFlags.StringVar(&prefixTag, "prefix-tag", "", "prefix tag")
 	cliFlags.StringVar(&suffixTag, "suffix-tag", "", "suffix tag")
 	cliFlags.BoolVar(&passingOnly, "passing-only", true, "passing only")
+	cliFlags.BoolVar(&withGatewayAPI, "with-gateway-api", false, "with gateway api")
 	cliFlags.StringVar(&deriveNamespace, "derive-namespace", "", "derive namespace")
 	flags.Merge(cliFlags, httpFlags.ClientFlags())
 
@@ -93,9 +96,11 @@ func main() {
 		log.Fatal().Err(err).Msgf("Error creating kube config (kubeconfig=%s)", kubeConfigFile)
 	}
 	kubeClient := kubernetes.NewForConfigOrDie(kubeConfig)
+	gatewayClient := gwapi.NewForConfigOrDie(kubeConfig)
 
 	k8s.SetTrustDomain(trustDomain)
 
+	connector.EnabledGatewayAPI(withGatewayAPI)
 	connector.SetSyncCloudNamespace(deriveNamespace)
 
 	// Initialize the generic Kubernetes event recorder and associate it with the fsm-consul-connector pod resource
@@ -124,7 +129,7 @@ func main() {
 		events.GenericEventRecorder().FatalEvent(err, events.InitializationError, "Error creating consul client")
 	}
 
-	sink := connector.NewSink(ctx, kubeClient)
+	sink := connector.NewSink(ctx, kubeClient, gatewayClient, fsmNamespace)
 	source := &consulConnector.Source{
 		ConsulClient: consulClient,
 		Domain:       trustDomain,
@@ -161,7 +166,7 @@ func main() {
 	}
 
 	// Start the global log level watcher that updates the log level dynamically
-	go k8s.WatchAndUpdateLogLevel(msgBroker, stop)
+	go connector.WatchMeshConfigUpdated(msgBroker, stop)
 
 	<-stop
 	log.Info().Msgf("Stopping fsm-consul-connector %s; %s; %s", version.Version, version.GitCommit, version.BuildDate)

diff --git a/cmd/fsm-connector/eureka/fsm-eureka-connector.go b/cmd/fsm-connector/eureka/fsm-eureka-connector.go
@@ -12,6 +12,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	gwapi "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned"
 
 	"github.com/hudl/fargo"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -49,6 +50,7 @@ var (
 	prefixTag         string
 	suffixTag         string
 	deriveNamespace   string
+	withGatewayAPI    bool
 
 	scheme = runtime.NewScheme()
 )
@@ -73,6 +75,7 @@ func init() {
 	cliFlags.StringVar(&prefixTag, "prefix-tag", "", "prefix tag")
 	cliFlags.StringVar(&suffixTag, "suffix-tag", "", "suffix tag")
 	cliFlags.BoolVar(&passingOnly, "passing-only", true, "passing only")
+	cliFlags.BoolVar(&withGatewayAPI, "with-gateway-api", false, "with gateway api")
 	cliFlags.StringVar(&deriveNamespace, "derive-namespace", "", "derive namespace")
 
 	_ = clientgoscheme.AddToScheme(scheme)
@@ -93,9 +96,11 @@ func main() {
 		log.Fatal().Err(err).Msgf("Error creating kube config (kubeconfig=%s)", kubeConfigFile)
 	}
 	kubeClient := kubernetes.NewForConfigOrDie(kubeConfig)
+	gatewayClient := gwapi.NewForConfigOrDie(kubeConfig)
 
 	k8s.SetTrustDomain(trustDomain)
 
+	connector.EnabledGatewayAPI(withGatewayAPI)
 	connector.SetSyncCloudNamespace(deriveNamespace)
 
 	// Initialize the generic Kubernetes event recorder and associate it with the fsm-eureka-connector pod resource
@@ -121,7 +126,7 @@ func main() {
 
 	eurekaClient := fargo.NewConn(httpAddr)
 
-	sink := connector.NewSink(ctx, kubeClient)
+	sink := connector.NewSink(ctx, kubeClient, gatewayClient, fsmNamespace)
 	source := &eurekaConnector.Source{
 		EurekaClient: &eurekaClient,
 		Domain:       trustDomain,
@@ -158,7 +163,7 @@ func main() {
 	}
 
 	// Start the global log level watcher that updates the log level dynamically
-	go k8s.WatchAndUpdateLogLevel(msgBroker, stop)
+	go connector.WatchMeshConfigUpdated(msgBroker, stop)
 
 	<-stop
 	log.Info().Msgf("Stopping fsm-eureka-connector %s; %s; %s", version.Version, version.GitCommit, version.BuildDate)

diff --git a/pkg/connector/consul/source.go b/pkg/connector/consul/source.go
@@ -88,7 +88,7 @@ func (s *Source) Run(ctx context.Context) {
 				services[connector.MicroSvcName(s.Prefix+name)] = connector.MicroSvcDomainName(fmt.Sprintf("%s.service.%s", name, s.Domain))
 			}
 		}
-		log.Info().Msgf("received services from Consul, count:%d", len(services))
+		log.Trace().Msgf("received services from Consul, count:%d", len(services))
 		s.Sink.SetServices(services)
 	}
 }
@@ -102,7 +102,7 @@ func (s *Source) Aggregate(svcName connector.MicroSvcName, svcDomainName connect
 		log.Err(err).Msgf("can't retrieve consul service, name:%s", string(svcName))
 		return nil, connector.ConsulDiscoveryService
 	}
-	log.Info().Msgf("PassingOnly:%v FilterTag:%v len(serviceEntries):%d", s.PassingOnly, s.FilterTag, len(serviceEntries))
+	log.Trace().Msgf("PassingOnly:[%v] FilterTag:[%v] len(serviceEntries):[%d]", s.PassingOnly, s.FilterTag, len(serviceEntries))
 	if len(serviceEntries) == 0 {
 		return nil, connector.ConsulDiscoveryService
 	}