Node agent mvp initial implementation (#65)

* node agent initial implementation

* pass NodeIdentifier in fornaxcore grpc GetMessage method

* a lots of node agent changes after integtest with fornaxcore

* create a integtest fornaxcore grpc server

* cleanup pods from store when it's terminated

* remove not used ContainerStore, run test in all target

Author: guowenjian90

Makefile

@@ -17,7 +17,7 @@ SHELL = /usr/bin/env bash -o pipefail
 .SHELLFLAGS = -ec
 
 .PHONY: all
-all: build
+all: build test
 
 ##@ General
 
@@ -40,11 +40,11 @@ help: ## Display this help.
 
 .PHONY: manifests
 manifests: controller-gen ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
-	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="./pkg/..." output:crd:artifacts:config=config/crd/bases
+	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="./pkg/apis/..." output:crd:artifacts:config=config/crd/bases
 
 .PHONY: generate
 generate: controller-gen openapi-gen client-gen ## generate-client-gen ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations.
-	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./pkg/..."
+	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./pkg/apis/core/..."
 	# $(OPENAPI_GEN) --go-header-file="hack/boilerplate.go.txt" --input-dirs="./pkg/apis/core/..." --output-package="centaurusinfra.io/fornax-serverless/pkg/apis/openapi"
 
 GENERATE_GROUPS = $(shell pwd)/hack/generate-groups.sh
@@ -71,6 +71,7 @@ test: manifests generate fmt vet envtest ## Run tests.
 build: generate fmt vet ## Build binary.
 	go build ./...
 	go build -o bin/apiserver cmd/apiserver/main.go
+	go build -o bin/nodeagent cmd/nodeagent/main.go
 
 .PHONY: run
 run: manifests generate fmt vet ## Run from your host.
@@ -155,6 +156,7 @@ protoc-gen: ## Download protc-gen locally if necessary.
 	$(call go-get-tool,$(PROTOC_GEN),google.golang.org/protobuf/cmd/[email protected])
 	$(call go-get-tool,$(PROTOC_GEN_GRPC),google.golang.org/grpc/cmd/[email protected])
 	$(call get-protoc,$(PROTOC))
+	go mod vendor
 	$(PROTOC) -I=./ -I=./vendor \
 		--go_out=../.. \
 		--go-grpc_out=../../ \

cmd/apiserver/app/grpc_server.go

@@ -0,0 +1,66 @@
+/*
+Copyright 2022.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package app
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	fornaxcore_grpc "centaurusinfra.io/fornax-serverless/pkg/fornaxcore/grpc"
+	"centaurusinfra.io/fornax-serverless/pkg/fornaxcore/grpc/server"
+	"centaurusinfra.io/fornax-serverless/pkg/fornaxcore/nodemonitor"
+	"centaurusinfra.io/fornax-serverless/pkg/fornaxcore/podmonitor"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"k8s.io/klog/v2"
+)
+
+func RunGrpcServer(ctx context.Context, port int, certFile, keyFile string) error {
+
+	lis, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%d", port))
+	if err != nil {
+		klog.ErrorS(err, "Fornaxcore grpc server failed to listen on port:", port)
+		return err
+	}
+	var opts []grpc.ServerOption
+	if certFile != "" && keyFile != "" {
+		creds, err := credentials.NewServerTLSFromFile(certFile, keyFile)
+		if err != nil {
+			klog.ErrorS(err, "Fornaxcore grpc server failed to generate credentials", "certFile", certFile, "keyFile", keyFile)
+			return err
+		}
+		opts = []grpc.ServerOption{grpc.Creds(creds)}
+	} else {
+
+	}
+
+	nodeMonitor := nodemonitor.New(10 * time.Second)
+	podMonitor := podmonitor.New(10 * time.Second)
+	server := server.New(nodeMonitor, podMonitor, nil)
+	grpcServer := grpc.NewServer(opts...)
+	fornaxcore_grpc.RegisterFornaxCoreServiceServer(grpcServer, server)
+	go func() {
+		err = grpcServer.Serve(lis)
+		if err != nil {
+			klog.ErrorS(err, "Fornaxcore grpc server stopped to serve")
+		}
+	}()
+
+	return nil
+}

cmd/apiserver/app/integtest_grpc_server.go

@@ -0,0 +1,66 @@
+/*
+Copyright 2022.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package app
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	fornaxcore_grpc "centaurusinfra.io/fornax-serverless/pkg/fornaxcore/grpc"
+	"centaurusinfra.io/fornax-serverless/pkg/fornaxcore/grpc/server"
+	"centaurusinfra.io/fornax-serverless/pkg/fornaxcore/nodemonitor"
+	"centaurusinfra.io/fornax-serverless/pkg/fornaxcore/podmonitor"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"k8s.io/klog/v2"
+)
+
+func RunIntegTestGrpcServer(ctx context.Context, port int, certFile, keyFile string) error {
+
+	lis, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%d", port))
+	if err != nil {
+		klog.ErrorS(err, "Fornaxcore grpc server failed to listen on port:", port)
+		return err
+	}
+	var opts []grpc.ServerOption
+	if certFile != "" && keyFile != "" {
+		creds, err := credentials.NewServerTLSFromFile(certFile, keyFile)
+		if err != nil {
+			klog.ErrorS(err, "Fornaxcore grpc server failed to generate credentials", "certFile", certFile, "keyFile", keyFile)
+			return err
+		}
+		opts = []grpc.ServerOption{grpc.Creds(creds)}
+	} else {
+
+	}
+
+	nodeMonitor := nodemonitor.NewIntegNodeMonitor(10 * time.Second)
+	podMonitor := podmonitor.NewIntegPodMonitor(10 * time.Second)
+	server := server.New(nodeMonitor, podMonitor, nil)
+	grpcServer := grpc.NewServer(opts...)
+	fornaxcore_grpc.RegisterFornaxCoreServiceServer(grpcServer, server)
+	go func() {
+		err = grpcServer.Serve(lis)
+		if err != nil {
+			klog.ErrorS(err, "Fornaxcore grpc server stopped to serve")
+		}
+	}()
+
+	return nil
+}

cmd/apiserver/main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -28,6 +29,7 @@ import (
 	"sigs.k8s.io/apiserver-runtime/pkg/builder"
 
 	// +kubebuilder:scaffold:resource-imports
+	"centaurusinfra.io/fornax-serverless/cmd/apiserver/app"
 	fornaxv1 "centaurusinfra.io/fornax-serverless/pkg/apis/core/v1"
 )
 
@@ -52,7 +54,21 @@ func main() {
 		}
 	}
 
-	err := builder.APIServer.
+	// start fornaxcore grpc server
+	klog.Info("starting fornaxcore grpc server")
+	port := 18001
+	certFile := ""
+	keyFile := ""
+	// err := app.RunGrpcServer(context.Background(), port, certFile, keyFile)
+	err := app.RunIntegTestGrpcServer(context.Background(), port, certFile, keyFile)
+	if err != nil {
+		klog.Fatal(err)
+	}
+	klog.Info("Fornaxcore grpc server started")
+
+	klog.Info("starting fornaxcore k8s.io rest api server")
+	// +kubebuilder:scaffold:resource-register
+	apiserver := builder.APIServer.
 		WithLocalDebugExtension().
 		WithResourceAndStorage(&fornaxv1.Application{}, storageFunc).
 		WithResourceAndStorage(&fornaxv1.ApplicationInstance{}, storageFunc).
@@ -63,9 +79,38 @@ func main() {
 			fmt.Printf("%T\n", config.Authentication.Authenticator)
 			fmt.Println(config.Authentication.APIAudiences)
 			return config
-		}).
-		Execute()
+		})
+
+	// WithResource(&v1.Pod{}).
+	// WithResource(&k8scorev1.FornaxNode{}).
+	// WithResource(&k8scorev1.FornaxSecret{}).
+	// WithResource(&k8scorev1.FornaxConfigMap{}).
+	// WithResource(&k8scorev1.FornaxServiceAccount{}).
+	// WithFlagFns(fns ...func(set *pflag.FlagSet) *pflag.FlagSet).
+	// // Authentication is the configuration for authentication
+	// Authentication AuthenticationInfo
+	//
+	// // Authorization is the configuration for authorization
+	// Authorization AuthorizationInfo
+
+	// LoopbackClientConfig is a config for a privileged loopback connection to the API server
+	// This is required for proper functioning of the PostStartHooks on a GenericAPIServer
+	// TODO: move into SecureServing(WithLoopback) as soon as insecure serving is gone
+	// LoopbackClientConfig * restclient.Config
+
+	// EgressSelector provides a lookup mechanism for dialing outbound connections.
+	// It does so based on a EgressSelectorConfiguration which was read at startup.
+	// EgressSelector * egressselector.EgressSelector
+
+	// WithServerFns(func(server *builder.GenericAPIServer) *builder.GenericAPIServer {
+	//  apiGroupInfo := *apiserver.APIGroupInfo{}
+	//  server.InstallLegacyAPIGroup(apiserver.DefaultLegacyAPIPrefix, apiGroupInfo)
+	//  return server
+	// }).
+
+	err = apiserver.Execute()
 	if err != nil {
 		klog.Fatal(err)
 	}
+
 }