Establishing deployer notifier connection (#183)

This PR creates a grpc connection between the deployer and notifier. The connection will later be used to send events.

cmd/deployer/app/register.go

@@ -61,7 +61,7 @@ func NewCompute(apiserverEp string, computeSpec openapi.ComputeSpec, bInsecure b
 	return compute, nil
 }
 
-func (compute *ComputeResource) RegisterNewCompute() error {
+func (compute *ComputeResource) RegisterNewCompute() (openapi.ComputeSpec, error) {
 	// construct URL
 	uriMap := map[string]string{}
 	url := restapi.CreateURL(compute.apiserverEp, restapi.RegisterComputeEndpoint, uriMap)
@@ -70,12 +70,12 @@ func (compute *ComputeResource) RegisterNewCompute() error {
 	if err != nil || restapi.CheckStatusCode(code) != nil {
 		zap.S().Errorf("Failed to register compute, sent computeSpec: %v, resp: %v, code: %d, err: %v",
 			compute.spec, string(resp), code, err)
-		return err
+		return openapi.ComputeSpec{}, err
 	}
 
 	zap.S().Infof("Success in registering new compute, sent obj: %v, resp: %v, code: %d",
 		compute.spec, string(resp), code)
 	compute.registered = true
 
-	return nil
+	return compute.spec, nil
 }

cmd/deployer/app/resourcehandler.go

@@ -0,0 +1,101 @@
+// Copyright 2022 Cisco Systems, Inc. and its affiliates
+//
+// 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.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package app
+
+import (
+	"crypto/tls"
+	"net/http"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/cisco-open/flame/pkg/openapi"
+)
+
+type resourceHandler struct {
+	apiserverEp string
+	notifierEp  string
+	spec        openapi.ComputeSpec
+
+	grpcDialOpt grpc.DialOption
+}
+
+func NewResourceHandler(apiserverEp string, notifierEp string, computeSpec openapi.ComputeSpec,
+	bInsecure bool, bPlain bool) *resourceHandler {
+	var grpcDialOpt grpc.DialOption
+
+	if bPlain {
+		grpcDialOpt = grpc.WithTransportCredentials(insecure.NewCredentials())
+	} else {
+		tlsCfg := &tls.Config{}
+		if bInsecure {
+			zap.S().Warn("Warning: allow insecure connection\n")
+
+			tlsCfg.InsecureSkipVerify = true
+			http.DefaultTransport.(*http.Transport).TLSClientConfig = tlsCfg
+		}
+		grpcDialOpt = grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg))
+	}
+
+	rHandler := &resourceHandler{
+		apiserverEp: apiserverEp,
+		notifierEp:  notifierEp,
+		spec:        computeSpec,
+		grpcDialOpt: grpcDialOpt,
+	}
+
+	return rHandler
+}
+
+// start connects to the notifier via grpc and handles notifications from the notifier
+func (r *resourceHandler) Start() {
+	go r.doStart()
+}
+
+func (r *resourceHandler) doStart() {
+	pauseTime := 10 * time.Second
+
+	for {
+		expBackoff := backoff.NewExponentialBackOff()
+		expBackoff.MaxElapsedTime = 5 * time.Minute // max wait time: 5 minutes
+		err := backoff.Retry(r.connect, expBackoff)
+		if err != nil {
+			zap.S().Fatalf("Cannot connect with notifier: %v", err)
+		}
+
+		// if connection is broken right after connection is made, this can cause
+		// too many connection/disconnection events. To migitage that, add some static
+		// pause time.
+		time.Sleep(pauseTime)
+	}
+}
+
+func (r *resourceHandler) connect() error {
+	// dial server
+	conn, err := grpc.Dial(r.notifierEp, r.grpcDialOpt)
+	if err != nil {
+		zap.S().Debugf("Cannot connect with notifier: %v, conn: %v", err, conn)
+		return err
+	}
+	zap.S().Infof("Connected with notifier at: %s", r.notifierEp)
+
+	return nil
+}

cmd/deployer/cmd/root.go

@@ -29,6 +29,7 @@ import (
 
 const (
 	argApiserver = "apiserver"
+	argNotifier  = "notifier"
 
 	optionInsecure = "insecure"
 	optionPlain    = "plain"
@@ -48,6 +49,14 @@ var rootCmd = &cobra.Command{
 			return fmt.Errorf("incorrect format for apiserver endpoint: %s", apiserver)
 		}
 
+		notifier, err := flags.GetString(argNotifier)
+		if err != nil {
+			return err
+		}
+		if len(strings.Split(notifier, ":")) != util.NumTokensInEndpoint {
+			return fmt.Errorf("incorrect format for notifier endpoint: %s", notifier)
+		}
+
 		bInsecure, _ := flags.GetBool(optionInsecure)
 		bPlain, _ := flags.GetBool(optionPlain)
 
@@ -69,11 +78,15 @@ var rootCmd = &cobra.Command{
 			return err
 		}
 
-		if err := compute.RegisterNewCompute(); err != nil {
+		computeSpec, err = compute.RegisterNewCompute()
+		if err != nil {
 			err = fmt.Errorf("error from RegisterNewCompute: %s", err)
 			return err
 		}
 
+		resoureHandler := app.NewResourceHandler(apiserver, notifier, computeSpec, bInsecure, bPlain)
+		resoureHandler.Start()
+
 		select {}
 	},
 }
@@ -83,6 +96,10 @@ func init() {
 	rootCmd.Flags().StringP(argApiserver, "a", defaultApiServerEp, "API server endpoint")
 	rootCmd.MarkFlagRequired(argApiserver)
 
+	defaultNotifierEp := fmt.Sprintf("0.0.0.0:%d", util.NotifierGrpcPort)
+	rootCmd.Flags().StringP(argNotifier, "n", defaultNotifierEp, "Notifier endpoint")
+	rootCmd.MarkFlagRequired(argNotifier)
+
 	rootCmd.PersistentFlags().Bool(optionInsecure, false, "Allow insecure connection")
 	rootCmd.PersistentFlags().Bool(optionPlain, false, "Allow unencrypted connection")
 }

fiab/helm-chart/templates/deployer-deployment.yaml

@@ -34,6 +34,8 @@ spec:
         - args:
           - --apiserver
           - "https://{{ .Values.frontDoorUrl.apiserver }}:443"
+          - --notifier
+          - "{{ .Values.frontDoorUrl.notifier }}:443"
           - --insecure
           command: ["/usr/bin/deployer"]
           image: {{ .Values.imageName }}:{{ .Values.imageTag }}