Add memory storage for supporting ListAndWatch

Makefile

@@ -48,7 +48,7 @@ ifeq ($(LATEST_TAG),$(shell git describe --abbrev=0 --tags))
 	VERSION=$(LATEST_TAG)
 endif
 
-all: apiserver clustersynchro-manager controller-manager
+all: apiserver binding-apiserver clustersynchro-manager controller-manager
 
 gen-clusterconfigs:
 	./hack/gen-clusterconfigs.sh
@@ -91,6 +91,13 @@ apiserver:
 			   -o bin/apiserver \
 			   cmd/apiserver/main.go
 
+.PHONY: binding-apiserver
+binding-apiserver:
+	CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH) go build \
+			   -ldflags "$(LDFLAGS)" \
+			   -o bin/binding-apiserver \
+			   cmd/binding-apiserver/main.go
+
 .PHONY: clustersynchro-manager
 clustersynchro-manager:
 	CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH) go build \
@@ -106,7 +113,7 @@ controller-manager:
 			   cmd/controller-manager/main.go
 
 .PHONY: images
-images: image-apiserver image-clustersynchro-manager image-controller-manager
+images: image-apiserver image-binding-apiserver image-clustersynchro-manager image-controller-manager
 
 image-apiserver:
 	GOOS="linux" $(MAKE) apiserver
@@ -116,7 +123,16 @@ image-apiserver:
 		--load \
 		--build-arg BASEIMAGE=$(BASEIMAGE) \
 		--build-arg BINNAME=apiserver .
-
+
+image-binding-apiserver:
+	GOOS="linux" $(MAKE) binding-apiserver
+	docker buildx build \
+		-t ${REGISTRY}/binding-apiserver-$(GOARCH):$(VERSION) \
+		--platform=linux/$(GOARCH) \
+		--load \
+		--build-arg BASEIMAGE=$(BASEIMAGE) \
+		--build-arg BINNAME=binding-apiserver .
+
 image-clustersynchro-manager: 
 	GOOS="linux" $(MAKE) clustersynchro-manager
 	docker buildx build \
@@ -136,7 +152,7 @@ image-controller-manager:
 		--build-arg BINNAME=controller-manager .
 
 .PHONY: push-images
-push-images: push-apiserver-image push-clustersynchro-manager-image push-controller-manager-image
+push-images: push-apiserver-image push-binding-apiserver-image push-clustersynchro-manager-image push-controller-manager-imag
 
 # clean manifest https://github.com/docker/cli/issues/954#issuecomment-586722447
 push-apiserver-image: clean-apiserver-manifest
@@ -160,6 +176,28 @@ push-apiserver-image: clean-apiserver-manifest
 		docker manifest push $(REGISTRY)/apiserver:latest; \
 	fi;
 
+# clean manifest https://github.com/docker/cli/issues/954#issuecomment-586722447
+push-binding-apiserver-image: clean-binding-apiserver-manifest
+	set -e; \
+	images=""; \
+	for arch in $(RELEASE_ARCHS); do \
+		GOARCH=$$arch $(MAKE) image-binding-apiserver; \
+		image=$(REGISTRY)/binding-apiserver-$$arch:$(VERSION); \
+		docker push $$image; \
+		images="$$images $$image"; \
+		if [ $(VERSION) != latest ]; then \
+			latest_image=$(REGISTRY)/binding-apiserver-$$arch:latest; \
+			docker tag $$image $$latest_image; \
+			docker push $$latest_image; \
+		fi; \
+	done; \
+	docker manifest create $(REGISTRY)/binding-apiserver:$(VERSION) $$images; \
+	docker manifest push $(REGISTRY)/binding-apiserver:$(VERSION); \
+	if [ $(VERSION) != latest ]; then \
+		docker manifest create $(REGISTRY)/binding-apiserver:latest $$images; \
+		docker manifest push $(REGISTRY)/binding-apiserver:latest; \
+	fi;
+
 # clean manifest https://github.com/docker/cli/issues/954#issuecomment-586722447
 push-clustersynchro-manager-image: clean-clustersynchro-manager-manifest
 	set -e; \
@@ -212,6 +250,10 @@ clean-apiserver-manifest:
 	docker manifest rm $(REGISTRY)/apiserver:$(VERSION) 2>/dev/null;\
 	docker manifest rm $(REGISTRY)/apiserver:latest 2>/dev/null; exit 0
 
+clean-binding-apiserver-manifest:
+	docker manifest rm $(REGISTRY)/binding-apiserver:$(VERSION) 2>/dev/null;\
+	docker manifest rm $(REGISTRY)/binding-apiserver:latest 2>/dev/null; exit 0
+
 clean-clustersynchro-manager-manifest:
 	docker manifest rm $(REGISTRY)/clustersynchro-manager:$(VERSION) 2>/dev/null;\
 	docker manifest rm $(REGISTRY)/clustersynchro-manager:latest 2>/dev/null; exit 0

cmd/apiserver/app/options/options.go

@@ -3,9 +3,12 @@ package options
 import (
 	"fmt"
 	"net"
+	"net/http"
+	"strings"
 
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apiserver/pkg/admission/plugin/namespace/lifecycle"
+	genericrequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	genericoptions "k8s.io/apiserver/pkg/server/options"
 	"k8s.io/apiserver/pkg/util/feature"
@@ -95,6 +98,12 @@ func (o *ClusterPediaServerOptions) Config() (*apiserver.Config, error) {
 	// genericConfig.OpenAPIConfig.Info.Title = openAPITitle
 	// genericConfig.OpenAPIConfig.Info.Version= openAPIVersion
 
+	// todo
+	// support watch to LongRunningFunc
+	genericConfig.LongRunningFunc = func(r *http.Request, requestInfo *genericrequest.RequestInfo) bool {
+		return strings.Contains(r.RequestURI, "watch")
+	}
+
 	if err := o.genericOptionsApplyTo(genericConfig); err != nil {
 		return nil, err
 	}

cmd/binding-apiserver/app/binding_apiserver.go

@@ -0,0 +1,99 @@
+package app
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	genericfeatures "k8s.io/apiserver/pkg/features"
+	cliflag "k8s.io/component-base/cli/flag"
+	"k8s.io/component-base/cli/globalflag"
+	"k8s.io/component-base/featuregate"
+	"k8s.io/component-base/logs"
+	"k8s.io/component-base/term"
+
+	"github.com/clusterpedia-io/clusterpedia/cmd/apiserver/app/options"
+	"github.com/clusterpedia-io/clusterpedia/pkg/generated/clientset/versioned"
+	"github.com/clusterpedia-io/clusterpedia/pkg/synchromanager"
+	clusterpediafeature "github.com/clusterpedia-io/clusterpedia/pkg/utils/feature"
+	"github.com/clusterpedia-io/clusterpedia/pkg/version/verflag"
+)
+
+func NewClusterPediaServerCommand(ctx context.Context) *cobra.Command {
+	opts := options.NewServerOptions()
+
+	cmd := &cobra.Command{
+		Use: "clusterpedia-apiserver",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			verflag.PrintAndExitIfRequested()
+
+			// Activate logging as soon as possible, after that
+			// show flags with the final logging configuration.
+			if err := opts.Logs.ValidateAndApply(clusterpediafeature.FeatureGate); err != nil {
+				return err
+			}
+			cliflag.PrintFlags(cmd.Flags())
+
+			config, err := opts.Config()
+			if err != nil {
+				return err
+			}
+
+			completedConfig := config.Complete()
+			if completedConfig.ClientConfig == nil {
+				return fmt.Errorf("CompletedConfig.New() called with config.ClientConfig == nil")
+			}
+			if completedConfig.StorageFactory == nil {
+				return fmt.Errorf("CompletedConfig.New() called with config.StorageFactory == nil")
+			}
+
+			crdclient, err := versioned.NewForConfig(completedConfig.ClientConfig)
+			if err != nil {
+				return err
+			}
+
+			synchromanager := synchromanager.NewManager(crdclient, config.StorageFactory)
+			go synchromanager.Run(1, ctx.Done())
+
+			server, err := completedConfig.New()
+			if err != nil {
+				return err
+			}
+
+			if err := server.Run(ctx); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+
+	namedFlagSets := opts.Flags()
+	verflag.AddFlags(namedFlagSets.FlagSet("global"))
+	globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name(), logs.SkipLoggingConfigurationFlags())
+	clusterpediafeature.MutableFeatureGate.AddFlag(namedFlagSets.FlagSet("mutable feature gate"))
+
+	fs := cmd.Flags()
+	for _, f := range namedFlagSets.FlagSets {
+		fs.AddFlagSet(f)
+	}
+
+	cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
+	cliflag.SetUsageAndHelpFunc(cmd, namedFlagSets, cols)
+	return cmd
+}
+
+func init() {
+	runtime.Must(logs.AddFeatureGates(clusterpediafeature.MutableFeatureGate))
+
+	// The feature gate `RemainingItemCount` should default to false
+	// https://github.com/clusterpedia-io/clusterpedia/issues/196
+	gates := clusterpediafeature.MutableFeatureGate.GetAll()
+	gate := gates[genericfeatures.RemainingItemCount]
+	gate.Default = false
+	gates[genericfeatures.RemainingItemCount] = gate
+
+	clusterpediafeature.MutableFeatureGate = featuregate.NewFeatureGate()
+	runtime.Must(clusterpediafeature.MutableFeatureGate.Add(gates))
+	clusterpediafeature.FeatureGate = clusterpediafeature.MutableFeatureGate
+}

cmd/binding-apiserver/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	"os"
+
+	apiserver "k8s.io/apiserver/pkg/server"
+	"k8s.io/component-base/cli"
+	_ "k8s.io/component-base/logs/json/register" // for JSON log format registration
+
+	"github.com/clusterpedia-io/clusterpedia/cmd/binding-apiserver/app"
+)
+
+func main() {
+	ctx := apiserver.SetupSignalContext()
+	command := app.NewClusterPediaServerCommand(ctx)
+	code := cli.Run(command)
+	os.Exit(code)
+}

deploy/binding-apiserver/clusterpedia_binding_apiserver_apiservice.yaml

@@ -0,0 +1,13 @@
+apiVersion: apiregistration.k8s.io/v1
+kind: APIService
+metadata:
+  name: v1beta1.clusterpedia.io
+spec:
+  insecureSkipTLSVerify: true
+  group: clusterpedia.io
+  groupPriorityMinimum: 1000
+  versionPriority: 100
+  service:
+    name: clusterpedia-binding-apiserver
+    namespace: clusterpedia-system
+  version: v1beta1

deploy/binding-apiserver/clusterpedia_binding_apiserver_deployment.yaml

@@ -0,0 +1,45 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: clusterpedia-binding-apiserver
+  namespace: clusterpedia-system
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: clusterpedia-binding-apiserver
+  namespace: clusterpedia-system
+spec:
+  ports:
+    - port: 443
+      protocol: TCP
+      targetPort: 443
+  selector:
+    app: clusterpedia-binding-apiserver
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: clusterpedia-binding-apiserver
+  namespace: clusterpedia-system
+  labels:
+    app: clusterpedia-binding-apiserver
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: clusterpedia-binding-apiserver
+  template:
+    metadata:
+      labels:
+        app: clusterpedia-binding-apiserver
+    spec:
+      containers:
+        - name: binding-apiserver
+          image: ghcr.io/clusterpedia-io/clusterpedia/binding-apiserver:v0.4.1
+          command:
+            - /usr/local/bin/binding-apiserver
+            - --secure-port=443
+            - --storage-name=memory
+            - -v=3
+      serviceAccountName: clusterpedia-binding-apiserver

deploy/binding-apiserver/clusterpedia_binding_apiserver_rbac.yaml

@@ -0,0 +1,24 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: clusterpedia
+rules:
+  - apiGroups: ['*']
+    resources: ['*']
+    verbs: ["*"]
+  - nonResourceURLs: ['*']
+    verbs: ["get"]
+
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: clusterpedia
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: clusterpedia
+subjects:
+  - kind: ServiceAccount
+    name: clusterpedia-binding-apiserver
+    namespace: clusterpedia-system

pkg/apiserver/apiserver.go

@@ -160,6 +160,7 @@ func (config completedConfig) New() (*ClusterPediaServer, error) {
 	genericServer.AddPostStartHookOrDie("start-clusterpedia-informers", func(context genericapiserver.PostStartHookContext) error {
 		clusterpediaInformerFactory.Start(context.StopCh)
 		clusterpediaInformerFactory.WaitForCacheSync(context.StopCh)
+
 		return nil
 	})
 

pkg/apiserver/registry/clusterpedia/collectionresources/rest.go

@@ -51,7 +51,7 @@ func NewREST(serializer runtime.NegotiatedSerializer, factory storage.StorageFac
 		for irt := range cr.ResourceTypes {
 			rt := &cr.ResourceTypes[irt]
 			if rt.Resource != "" {
-				config, err := configFactory.NewConfig(rt.GroupResource().WithVersion(""))
+				config, err := configFactory.NewConfig(rt.GroupResource().WithVersion(""), false)
 				if err != nil {
 					continue
 				}

pkg/kubeapiserver/apiserver.go

@@ -35,6 +35,7 @@ func init() {
 		&metav1.APIGroupList{},
 		&metav1.APIGroup{},
 		&metav1.APIResourceList{},
+		&metav1.WatchEvent{},
 	)
 }
 

pkg/kubeapiserver/resource_handler.go

@@ -118,6 +118,8 @@ func (r *ResourceHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		handler = handlers.GetResource(storage, reqScope)
 	case "list":
 		handler = handlers.ListResource(storage, nil, reqScope, false, r.minRequestTimeout)
+	case "watch":
+		handler = handlers.ListResource(storage, storage, reqScope, true, r.minRequestTimeout)
 	default:
 		responsewriters.ErrorNegotiated(
 			apierrors.NewMethodNotSupported(gvr.GroupResource(), requestInfo.Verb),

pkg/kubeapiserver/resourcerest/storage.go

@@ -9,6 +9,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/watch"
 	genericrequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericfeatures "k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/registry/rest"
@@ -38,6 +39,7 @@ type RESTStorage struct {
 
 var _ rest.Lister = &RESTStorage{}
 var _ rest.Getter = &RESTStorage{}
+var _ rest.Watcher = &RESTStorage{}
 
 func (s *RESTStorage) New() runtime.Object {
 	return s.NewFunc()
@@ -121,3 +123,7 @@ func (s *RESTStorage) ConvertToTable(ctx context.Context, object runtime.Object,
 
 	return printers.NewDefaultTableConvertor(s.DefaultQualifiedResource).ConvertToTable(ctx, object, tableOptions)
 }
+
+func (s *RESTStorage) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
+	return s.Storage.Watch(ctx, options)
+}