Support for north-south traffic (#835)

* Support for north-south traffic

* fix ci

* fix ci

Author: mfordjody

dubbod/planet/cmd/planet-agent/app/cmd.go

@@ -21,9 +21,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/apache/dubbo-kubernetes/pkg/log"
 	"net/netip"
 
+	"github.com/apache/dubbo-kubernetes/pkg/log"
+
 	"github.com/apache/dubbo-kubernetes/dubbod/planet/cmd/planet-agent/options"
 	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/util/network"
 	"github.com/apache/dubbo-kubernetes/pkg/cmd"

dubbod/planet/docker/dockerfile.proxyadapter

@@ -15,5 +15,6 @@
 
 FROM gcr.io/distroless/static:debug
 COPY bin/planet-agent /usr/local/bin/planet-agent
+COPY bin/pixiugateway /usr/local/bin/pixiugateway
 USER 9999:9999
 ENTRYPOINT ["/usr/local/bin/planet-agent"]

…d/planet/pkg/bootstrap/certcontroller.go …/planet/pkg/bootstrap/cert_controller.godubbod/planet/pkg/bootstrap/certcontroller.go renamed to dubbod/planet/pkg/bootstrap/cert_controller.go dubbod/planet/pkg/bootstrap/certcontroller.go renamed to dubbod/planet/pkg/bootstrap/cert_controller.go

@@ -21,9 +21,16 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/json"
 	"encoding/pem"
 	"fmt"
 	configaggregate "github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/config/aggregate"
+	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/config/kube/gateway"
+	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/features"
+	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/leaderelection"
+	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/leaderelection/k8sleaderelection/k8sresourcelock"
+	"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvr"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"net/url"
 	"strings"
 
@@ -59,7 +66,9 @@ func (s *Server) makeKubeConfigController(args *PlanetArgs) *crdclient.Client {
 	}
 
 	schemas := collections.Planet
-
+	if features.EnableGatewayAPI {
+		schemas = collections.PlanetGatewayAPI()
+	}
 	return crdclient.NewForSchemas(s.kubeClient, opts, schemas)
 }
 
@@ -69,6 +78,72 @@ func (s *Server) initK8SConfigStore(args *PlanetArgs) error {
 	}
 	configController := s.makeKubeConfigController(args)
 	s.ConfigStores = append(s.ConfigStores, configController)
+
+	if features.EnableGatewayAPI {
+		if s.statusManager == nil && features.EnableGatewayAPIStatus {
+			s.initStatusManager(args)
+		}
+		args.RegistryOptions.KubeOptions.KrtDebugger = args.KrtDebugger
+		gwc := gateway.NewController(s.kubeClient, s.kubeClient.CrdWatcher().WaitForCRD, args.RegistryOptions.KubeOptions, s.XDSServer)
+		s.environment.GatewayAPIController = gwc
+		s.ConfigStores = append(s.ConfigStores, s.environment.GatewayAPIController)
+
+		// Use a channel to signal activation of per-revision status writer
+		activatePerRevisionStatusWriterCh := make(chan struct{})
+		s.checkAndRunNonRevisionLeaderElectionIfRequired(args, activatePerRevisionStatusWriterCh)
+
+		s.addTerminatingStartFunc("gateway status", func(stop <-chan struct{}) error {
+			leaderelection.
+				NewPerRevisionLeaderElection(args.Namespace, args.PodName, leaderelection.GatewayStatusController, args.Revision, s.kubeClient).
+				AddRunFunction(func(leaderStop <-chan struct{}) {
+					log.Infof("waiting for gateway status writer activation")
+					<-activatePerRevisionStatusWriterCh
+					log.Infof("Starting gateway status writer for revision: %s", args.Revision)
+					gwc.SetStatusWrite(true, s.statusManager)
+
+					// Trigger a push so we can recompute status
+					s.XDSServer.ConfigUpdate(&model.PushRequest{
+						Full:   true,
+						Reason: model.NewReasonStats(model.GlobalUpdate),
+						Forced: true,
+					})
+					<-leaderStop
+					log.Infof("Stopping gateway status writer")
+					gwc.SetStatusWrite(false, nil)
+				}).
+				Run(stop)
+			return nil
+		})
+		if features.EnableGatewayAPIDeploymentController {
+			s.addTerminatingStartFunc("gateway deployment controller", func(stop <-chan struct{}) error {
+				leaderelection.
+					NewPerRevisionLeaderElection(args.Namespace, args.PodName, leaderelection.GatewayDeploymentController, args.Revision, s.kubeClient).
+					AddRunFunction(func(leaderStop <-chan struct{}) {
+						// We can only run this if the Gateway CRD is created
+						if s.kubeClient.CrdWatcher().WaitForCRD(gvr.KubernetesGateway, leaderStop) {
+							controller := gateway.NewDeploymentController(s.kubeClient, s.clusterID, s.environment,
+								s.webhookInfo.getWebhookConfig, s.webhookInfo.addHandler, nil, args.Revision, args.Namespace)
+							// Start informers again. This fixes the case where informers for namespace do not start,
+							// as we create them only after acquiring the leader lock
+							// Note: stop here should be the overall pilot stop, NOT the leader election stop. We are
+							// basically lazy loading the informer, if we stop it when we lose the lock we will never
+							// recreate it again.
+							s.kubeClient.RunAndWait(stop)
+							// TODO tag watcher
+							controller.Run(leaderStop)
+						}
+					}).
+					Run(stop)
+				return nil
+			})
+		}
+	}
+	var err error
+	s.RWConfigStore, err = configaggregate.MakeWriteableCache(s.ConfigStores, configController)
+	if err != nil {
+		return err
+	}
+
 	s.XDSServer.ConfigUpdate(&model.PushRequest{
 		Full:   true,
 		Reason: model.NewReasonStats(model.GlobalUpdate),
@@ -181,9 +256,6 @@ func (s *Server) initConfigController(args *PlanetArgs) error {
 		}
 	}
 
-	// TODO ingress controller
-	// TODO addTerminatingStartFunc
-
 	// Wrap the config controller with a cache.
 	aggregateConfigController, err := configaggregate.MakeCache(s.ConfigStores)
 	if err != nil {
@@ -298,3 +370,60 @@ func (s *Server) getRootCertFromSecret(name, namespace string) (*dubboCredential
 	}
 	return kube.ExtractRoot(secret.Data)
 }
+
+func (s *Server) checkAndRunNonRevisionLeaderElectionIfRequired(args *PlanetArgs, activateCh chan struct{}) {
+	cm, err := s.kubeClient.Kube().CoreV1().ConfigMaps(args.Namespace).Get(context.Background(), leaderelection.GatewayStatusController, v1.GetOptions{})
+
+	if errors.IsNotFound(err) {
+		// ConfigMap does not exist, so per-revision leader election should be active
+		close(activateCh)
+		return
+	}
+	leaderAnn, ok := cm.Annotations[k8sresourcelock.LeaderElectionRecordAnnotationKey]
+	if ok {
+		var leaderInfo struct {
+			HolderIdentity string `json:"holderIdentity"`
+		}
+		if err := json.Unmarshal([]byte(leaderAnn), &leaderInfo); err == nil {
+			if leaderInfo.HolderIdentity != "" {
+				// Non-revision leader election should run, per-revision should be waiting for activation
+				s.addTerminatingStartFunc("gateway status", func(stop <-chan struct{}) error {
+					secondStop := make(chan struct{})
+					// if stop closes, ensure secondStop closes too
+					go func() {
+						<-stop
+						select {
+						case <-secondStop:
+						default:
+							close(secondStop)
+						}
+					}()
+					leaderelection.
+						NewLeaderElection(args.Namespace, args.PodName, leaderelection.GatewayStatusController, args.Revision, s.kubeClient).
+						AddRunFunction(func(leaderStop <-chan struct{}) {
+							// now that we have the leader lock, we can activate the per-revision status writer
+							// first close the activateCh channel if it is not already closed
+							log.Infof("Activating gateway status writer")
+							select {
+							case <-activateCh:
+								// Channel already closed, do nothing
+							default:
+								close(activateCh)
+							}
+							// now end this lease itself
+							select {
+							case <-secondStop:
+							default:
+								close(secondStop)
+							}
+						}).
+						Run(secondStop)
+					return nil
+				})
+				return
+			}
+		}
+	}
+	// If annotation missing or holderIdentity is blank, per-revision leader election should be active
+	close(activateCh)
+}

…planet/pkg/bootstrap/configcontroller.go …lanet/pkg/bootstrap/config_controller.godubbod/planet/pkg/bootstrap/configcontroller.go renamed to dubbod/planet/pkg/bootstrap/config_controller.go dubbod/planet/pkg/bootstrap/configcontroller.go renamed to dubbod/planet/pkg/bootstrap/config_controller.go

@@ -29,6 +29,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/status"
+
 	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/features"
 	dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/grpc"
 	"github.com/apache/dubbo-kubernetes/dubbod/planet/pkg/keycertbundle"
@@ -127,6 +129,9 @@ type Server struct {
 	webhookInfo *webhookInfo
 
 	krtDebugger *krt.DebugHandler
+
+	RWConfigStore model.ConfigStoreController
+	statusManager *status.Manager
 }
 
 type readinessFlags struct {
@@ -139,6 +144,28 @@ type webhookInfo struct {
 	wh *inject.Webhook
 }
 
+func (w *webhookInfo) getWebhookConfig() inject.Config {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+	if w.wh != nil && w.wh.Config != nil {
+		return *w.wh.Config
+	}
+	return inject.Config{}
+}
+
+func (w *webhookInfo) addHandler(fn func()) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	// Note: Dubbo's Webhook doesn't have RegisterInjectionHandler method
+	// This is a placeholder for future implementation
+	// For now, we'll call the handler directly if webhook is available
+	if w.wh != nil {
+		// Handler will be called when webhook config changes
+		// This is a simplified implementation compared to Istio
+		fn()
+	}
+}
+
 type readinessProbe func() bool
 
 func NewServer(args *PlanetArgs, initFuncs ...func(*Server)) (*Server, error) {
@@ -448,7 +475,11 @@ func (s *Server) initRegistryEventHandlers() {
 
 		// Build ConfigKey for the changed config
 		// Find the schema to get the kind.Kind
+		// First try Planet schemas, then try PlanetGatewayAPI schemas
 		schema, found := collections.Planet.FindByGroupVersionKind(cfg.GroupVersionKind)
+		if !found && features.EnableGatewayAPI {
+			schema, found = collections.PlanetGatewayAPI().FindByGroupVersionKind(cfg.GroupVersionKind)
+		}
 		if !found {
 			log.Warnf("configHandler: schema not found for %v, skipping", cfg.GroupVersionKind)
 			return
@@ -468,6 +499,12 @@ func (s *Server) initRegistryEventHandlers() {
 			configKind = kind.ServiceRoute
 		case "PeerAuthentication":
 			configKind = kind.PeerAuthentication
+		case "GatewayClass":
+			configKind = kind.GatewayClass
+		case "Gateway":
+			configKind = kind.Gateway
+		case "HTTPRoute":
+			configKind = kind.HTTPRoute
 		default:
 			log.Debugf("configHandler: unknown schema identifier %s for %v, skipping", schemaID, cfg.GroupVersionKind)
 			return
@@ -482,11 +519,12 @@ func (s *Server) initRegistryEventHandlers() {
 		// Log the config change
 		log.Infof("configHandler: %s event for %s/%s/%s", event, configKey.Kind, configKey.Namespace, configKey.Name)
 
-		// Some configs (SubsetRule/ServiceRoute/PeerAuthentication) require Full push to ensure
+		// Some configs (SubsetRule/ServiceRoute/PeerAuthentication/HTTPRoute) require Full push to ensure
 		// PushContext is re-initialized and configuration is reloaded.
 		// PeerAuthentication must rebuild AuthenticationPolicies to enable STRICT mTLS on LDS; without
 		// a full push the cached PushContext would continue serving plaintext listeners.
-		needsFullPush := configKind == kind.SubsetRule || configKind == kind.ServiceRoute || configKind == kind.PeerAuthentication
+		// HTTPRoute must rebuild HTTPRoute index to enable Gateway API routing.
+		needsFullPush := configKind == kind.SubsetRule || configKind == kind.ServiceRoute || configKind == kind.PeerAuthentication || configKind == kind.HTTPRoute
 
 		// Trigger ConfigUpdate to push changes to all connected proxies
 		s.XDSServer.ConfigUpdate(&model.PushRequest{
@@ -496,6 +534,9 @@ func (s *Server) initRegistryEventHandlers() {
 		})
 	}
 	schemas := collections.Planet.All()
+	if features.EnableGatewayAPI {
+		schemas = collections.PlanetGatewayAPI().All()
+	}
 	log.Debugf("initRegistryEventHandlers: found %d schemas to register", len(schemas))
 	registeredCount := 0
 	for _, schema := range schemas {
@@ -951,6 +992,14 @@ func (s *Server) shouldStartNsController() bool {
 	return true
 }
 
+func (s *Server) initStatusManager(_ *PlanetArgs) {
+	s.addStartFunc("status manager", func(stop <-chan struct{}) error {
+		s.statusManager = status.NewManager(s.RWConfigStore)
+		s.statusManager.Start(stop)
+		return nil
+	})
+}
+
 func getDNSNames(_ *PlanetArgs, host string) []string {
 	// Append custom hostname if there is any
 	customHost := features.DubbodServiceCustomHost
@@ -1054,3 +1103,7 @@ func (s *Server) reloadDubbodCert(watchCh <-chan struct{}, stopCh <-chan struct{
 		}
 	}
 }
+
+func (s *Server) addTerminatingStartFunc(name string, fn server.Component) {
+	s.server.RunComponentAsyncAndWait(name, fn)
+}