diff --git a/cmd/main.go b/cmd/main.go index 6639a30..7f8c468 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -59,3 +59,4 @@ func main() { exitCode = -1 } } + diff --git a/example/client/deployments/helm/xds-grpc-client-example/values.yaml b/example/client/deployments/helm/xds-grpc-client-example/values.yaml index 4bc0160..f338638 100644 --- a/example/client/deployments/helm/xds-grpc-client-example/values.yaml +++ b/example/client/deployments/helm/xds-grpc-client-example/values.yaml @@ -54,7 +54,7 @@ configMap: LOGGER_LEVEL: "info" GRPC_GO_LOG_VERBOSITY_LEVEL: 99 GRPC_GO_LOG_SEVERITY_LEVEL: "info" - Server1Address: "xds:///xds-grpc-server-example-headless:8888" + Server1Address: "xds:///xds-grpc-server-example-headless.control-plane-example:8888" GRPC_XDS_BOOTSTRAP: "./xds_bootstrap.json" # GRPC_XDS_BOOTSTRAP_CONFIG: "{\n \"xds_servers\": [\n {\n \"server_uri\": \"xds-control-plane-headless.xds-control-plane.svc.cluster.local:8888\",\n \"channel_creds\": [\n {\n \"type\": \"insecure\"\n }\n ],\n \"server_features\": [\"xds_v3\"] \n }\n ],\n \"node\": {\n \"id\": \"b7f9c818-fb46-43ca-8662-d3bdbcf7ec18~10.0.0.1\",\n \"metadata\": {\n \"R_GCP_PROJECT_NUMBER\": \"123456789012\"\n },\n \"locality\": {\n \"zone\": \"us-central1-a\"\n }\n }\n}\n" diff --git a/example/client/main.go b/example/client/main.go index 0464a6f..584db61 100644 --- a/example/client/main.go +++ b/example/client/main.go @@ -43,7 +43,7 @@ var ( func main() { // viper.SetDefault("Server1Address", "xds:///xds-grpc-server-example-headless:8888") - viper.SetDefault("Server1Address", "xds:///xds-grpc-server-example-headless") + viper.SetDefault("Server1Address", "xds:///xds-grpc-server-example-headless.control-plane-example") // viper.SetDefault("Server1Address", "xds-grpc-server-example-headless:8888") // Read Config from ENV viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) diff --git a/internal/app/control-plane/controller.go b/internal/app/control-plane/controller.go deleted file mode 100644 index 145ce9e..0000000 --- a/internal/app/control-plane/controller.go +++ /dev/null @@ -1 +0,0 @@ -package controlplane diff --git a/internal/app/control-plane/informer.go b/internal/app/control-plane/informer.go index 1f284a9..f930349 100644 --- a/internal/app/control-plane/informer.go +++ b/internal/app/control-plane/informer.go @@ -35,8 +35,8 @@ func (a *App) OnDeleteService(key string, serviceObj *v1.Service) { } func (a *App) OnUpdateService(newKey string, newServiceObj *v1.Service, oldKey string, oldServiceObj *v1.Service) { - slog.Info("OnUpdateService", "newKey", newKey, "newServiceName", newServiceObj.Name, "newServiceNamespace", newServiceObj.Namespace, - "oldKey", oldKey, "oleServiceName", oldServiceObj.Name, "oldServiceNamespace", oldServiceObj.Namespace) + // slog.Info("OnUpdateService", "newKey", newKey, "newServiceName", newServiceObj.Name, "newServiceNamespace", newServiceObj.Namespace, + // "oldKey", oldKey, "oleServiceName", oldServiceObj.Name, "oldServiceNamespace", oldServiceObj.Namespace) a.muResource.Lock() defer a.muResource.Unlock() resourceInstance, ok := a.resources[oldKey] @@ -51,3 +51,6 @@ func (a *App) OnUpdateService(newKey string, newServiceObj *v1.Service, oldKey s resourceInstance.ServiceObj = newServiceObj a.resources[newKey] = resourceInstance } + + +// xds-grpc-server-example-headless.control-plane-example \ No newline at end of file diff --git a/internal/app/control-plane/xds.go b/internal/app/control-plane/xds.go index 49ef662..b659487 100644 --- a/internal/app/control-plane/xds.go +++ b/internal/app/control-plane/xds.go @@ -10,12 +10,20 @@ import ( // *** callbacks func (a *App) NewStreamRequest(id string, resourceNames []string, typeURL string) { - node := a.CreateNode(id) + isUpdateNeeded := false + node, err := a.GetNode(id) + if err != nil { + node = a.CreateNode(id) + isUpdateNeeded = true + } slog.Info("app NewStreamRequest", "id", id, "resourceNames", resourceNames, "typeURL", typeURL) for _, rn := range resourceNames { node.AddWatching(rn) a.AddResourceWatchToNode(id, rn, typeURL) } + if isUpdateNeeded { + a.UpdateNodeCache(id) + } } func (a *App) StreamClosed(id string) { @@ -34,7 +42,7 @@ func (a *App) UpdateNodeCache(nodeID string) { } resources := node.GetWatchings() node.ClearResources() - slog.Info("UpdateCache", "nodeID", nodeID) + slog.Info("UpdateCache", "nodeID", nodeID, "version", node.GetVersion()) for _, rn := range resources { resource, ok := a.resources[rn] if !ok { @@ -42,11 +50,15 @@ func (a *App) UpdateNodeCache(nodeID string) { continue } //ToDo: later fix loop through each port name - endPoint, cluster, listner, route, err := xds.MakeXDSResource(resource, a.conf.Region, a.conf.Zone, resource.ServiceObj.Spec.Ports[0].Name) + endPoint, cluster, route, listner, err := xds.MakeXDSResource(resource, a.conf.Region, a.conf.Zone, resource.ServiceObj.Spec.Ports[0].Name) if err != nil { slog.Error("UpdateCache, failed to Make XDS Resource", "error", err, "resource", rn, "nodeID", nodeID) continue } + slog.Info("UpdateCache, resource", "resource", rn, "nodeID", nodeID, "endPoint", endPoint) + slog.Info("UpdateCache, resource", "resource", rn, "nodeID", nodeID, "cluster", cluster) + slog.Info("UpdateCache, resource", "resource", rn, "nodeID", nodeID, "listner", listner) + slog.Info("UpdateCache, resource", "resource", rn, "nodeID", nodeID, "route",route) node.AddCluster(cluster) node.AddListener(listner) node.AddEndpoint(endPoint) diff --git a/internal/informer/service.go b/internal/informer/service.go index 42fdbd6..2f4cf9d 100644 --- a/internal/informer/service.go +++ b/internal/informer/service.go @@ -33,7 +33,7 @@ func NewServiceInformer(factory informers.SharedInformerFactory, handler Service } func getServiceKey(service *v1.Service) string { - return service.Namespace + "." + service.Name + return service.Name + "." + service.Namespace } func (si *ServiceInformer) Run(stopCh <-chan struct{}) {