Registration server pointing to LB or DP directly

Heartbeat logging bugfixes

Signed-off-by: Lazar Cvetković <[email protected]>

api/control_plane_api_server.go

@@ -56,7 +56,7 @@ func CreateNewCpApiServer(args *CpApiServerCreationArguments) (*CpApiServer, cha
 	cpApiServer := &CpApiServer{
 		LeaderElectionServer: leaderElectionServer,
 		ControlPlane:         cp,
-		HAProxyAPI:           haproxy.NewHAProxyAPI(),
+		HAProxyAPI:           haproxy.NewHAProxyAPI(args.Cfg.LoadBalancerAddress),
 	}
 
 	go grpc_helpers.CreateGRPCServer(args.Cfg.Port, func(sr grpc.ServiceRegistrar) {
@@ -218,7 +218,7 @@ func (c *CpApiServer) RegisterDataplane(ctx context.Context, in *proto.Dataplane
 	}
 
 	status, err := c.ControlPlane.RegisterDataplane(ctx, in)
-	if status.Success && err != nil {
+	if status.Success && err == nil {
 		c.HAProxyAPI.AddDataplane(in.IP, int(in.ProxyPort))
 	}
 
@@ -238,7 +238,7 @@ func (c *CpApiServer) DeregisterDataplane(ctx context.Context, in *proto.Datapla
 	}
 
 	status, err := c.ControlPlane.DeregisterDataplane(ctx, in)
-	if status.Success && err != nil {
+	if status.Success && err == nil {
 		c.HAProxyAPI.RemoveDataplane(in.IP, int(in.ProxyPort))
 	}
 
@@ -297,5 +297,6 @@ func (c *CpApiServer) AppendEntries(_ context.Context, args *proto.AppendEntries
 }
 
 func (c *CpApiServer) ReviseHAProxyServers() {
+	logrus.Infof("Revising HAProxy backend server list...")
 	c.ControlPlane.ReviseDataplanesInLB(c.HAProxyAPI.ReviseDataplanes)
 }

cmd/master_node/config.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: false
 reconstruct: false
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_cluster.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: false
 reconstruct: false
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_cluster_raft_1.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: true
 reconstruct: true
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_cluster_raft_2.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: true
 reconstruct: true
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_cluster_raft_3.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: true
 reconstruct: true
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_raft_1.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: true
 reconstruct: true
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_raft_2.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: true
 reconstruct: true
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

cmd/master_node/config_raft_3.yaml

@@ -7,6 +7,8 @@ placementPolicy: "random"
 persistence: true
 reconstruct: true
 
+loadBalancerAddress: ""
+
 removeWorkerNode: false
 removeDataplane: false
 

configs/haproxy.cfg

@@ -22,14 +22,14 @@ defaults
     timeout check           10s
 
 frontend dirigent
-    bind *:8080
-    mode tcp
+    bind *:8079
+    mode http
     option tcplog
     default_backend dirigent_data_planes
 
 backend dirigent_data_planes
     option httpchk GET /health
     http-check expect status 200
-    mode tcp
+    mode http
     balance hdr(host)
-        server dp_1 127.0.0.1:8080 check
+    #server dp_1 127.0.0.1:8080 check

internal/control_plane/registration_server/registration_server.go

@@ -124,10 +124,24 @@ func registrationHandler(cpApi *api.CpApiServer) func(w http.ResponseWriter, r *
 			return
 		}
 
-		endpointList := ""
+		_, err = w.Write([]byte(GetLoadBalancerAddress(cpApi)))
+		if err != nil {
+			http.Error(w, "Error writing endpoints.", http.StatusInternalServerError)
+			return
+		}
 
+		logrus.Debugf("Successfully registered function %s.", name)
+	}
+}
+
+func GetLoadBalancerAddress(cpApi *api.CpApiServer) string {
+	if cpApi.HAProxyAPI.GetLoadBalancerAddress() == "" {
+		endpointList := ""
 		cnt := 0
+
 		cpApi.ControlPlane.DataPlaneConnections.RLock()
+		defer cpApi.ControlPlane.DataPlaneConnections.RUnlock()
+
 		for _, conn := range cpApi.ControlPlane.DataPlaneConnections.GetMap() {
 			setDelimiter := cnt != cpApi.ControlPlane.DataPlaneConnections.Len()-1
 			delimiter := ""
@@ -137,18 +151,12 @@ func registrationHandler(cpApi *api.CpApiServer) func(w http.ResponseWriter, r *
 			}
 
 			endpointList += fmt.Sprintf("%s:%s%s", conn.GetIP(), conn.GetProxyPort(), delimiter)
-
 			cnt++
 		}
-		cpApi.ControlPlane.DataPlaneConnections.RUnlock()
 
-		_, err = w.Write([]byte(endpointList))
-		if err != nil {
-			http.Error(w, "Error writing endpoints.", http.StatusInternalServerError)
-			return
-		}
-
-		logrus.Debugf("Successfully registered function %s.", name)
+		return endpointList
+	} else {
+		return cpApi.HAProxyAPI.GetLoadBalancerAddress()
 	}
 }
 

internal/data_plane/haproxy/manager.go

@@ -17,19 +17,22 @@ const (
 )
 
 type API struct {
-	client *clientnative.HAProxyClient
+	client    *clientnative.HAProxyClient
+	lbAddress string
 
 	// addressTo - e.g., 10.0.1.2:8080 -> dataplane_123
 	addressToName map[string]string
 	mutex         sync.Mutex
 }
 
-func NewHAProxyAPI() *API {
+func NewHAProxyAPI(loadBalancerAddress string) *API {
 	config := getConfigClient()
 	runtime := getRuntimeClient(config)
 
 	api := &API{
-		client:        getHAProxyClient(config, runtime),
+		client:    getHAProxyClient(config, runtime),
+		lbAddress: loadBalancerAddress,
+
 		addressToName: make(map[string]string),
 	}
 
@@ -42,6 +45,10 @@ func NewHAProxyAPI() *API {
 	return api
 }
 
+func (api *API) GetLoadBalancerAddress() string {
+	return api.lbAddress
+}
+
 func (api *API) ListDataplanes() (names []string, addresses []string) {
 	_, servers, err := api.client.Configuration.GetServers(BackendName, "")
 	if err != nil {
@@ -84,11 +91,11 @@ func (api *API) addServer(ipAddress string, port int, transaction string, versio
 	defer api.mutex.Unlock()
 
 	if _, ok := api.addressToName[url]; ok {
-		logrus.Errorf("A server with the same network address already exists.")
 		return
 	} else {
 		api.persistServerMetadata(name, ipAddress, port, transaction, version)
 		api.addressToName[url] = name
+		logrus.Infof("Added data plane with address %s:%d to HAProxy backend.", ipAddress, port)
 	}
 }
 
@@ -116,7 +123,6 @@ func (api *API) removeServerByName(ipAddress string, port int, transaction strin
 	defer api.mutex.Unlock()
 
 	if dataplaneName, ok := api.addressToName[url]; !ok {
-		logrus.Errorf("Cannot delete requested server as it does not exist.")
 		return
 	} else {
 		err := api.client.Configuration.DeleteServer(dataplaneName, BackendName, transaction, version)
@@ -125,6 +131,7 @@ func (api *API) removeServerByName(ipAddress string, port int, transaction strin
 		}
 
 		delete(api.addressToName, url)
+		logrus.Infof("Removed data plane with address %s:%d to HAProxy backend.", ipAddress, port)
 	}
 }
 

internal/data_plane/haproxy/manager_test.go

@@ -5,7 +5,9 @@ import (
 )
 
 func TestEditBackend(t *testing.T) {
-	api := NewHAProxyAPI()
+	// To run this test make sure HAProxy is installed on the machine and
+	// configs/haproxy.cfg is placed in /etc/haproxy/ folder
+	api := NewHAProxyAPI("dummy")
 
 	api.DeleteAllDataplanes()
 	if k, _ := api.ListDataplanes(); len(k) != 0 {

pkg/config/config.go

@@ -10,19 +10,20 @@ import (
 )
 
 type ControlPlaneConfig struct {
-	Port               string         `mapstructure:"port"`
-	Replicas           []string       `mapstructure:"replicas"`
-	PortRegistration   string         `mapstructure:"portRegistration"`
-	Verbosity          string         `mapstructure:"verbosity"`
-	TraceOutputFolder  string         `mapstructure:"traceOutputFolder"`
-	PlacementPolicy    string         `mapstructure:"placementPolicy"`
-	Persistence        bool           `mapstructure:"persistence"`
-	Profiler           ProfilerConfig `mapstructure:"profiler"`
-	RedisConf          RedisConf      `mapstructure:"redis"`
-	Reconstruct        bool           `mapstructure:"reconstruct"`
-	RemoveWorkerNode   bool           `mapstructure:"removeWorkerNode"`
-	RemoveDataplane    bool           `mapstructure:"removeDataplane"`
-	PrecreateSnapshots bool           `mapstructure:"precreateSnapshots"`
+	Port                string         `mapstructure:"port"`
+	Replicas            []string       `mapstructure:"replicas"`
+	PortRegistration    string         `mapstructure:"portRegistration"`
+	Verbosity           string         `mapstructure:"verbosity"`
+	TraceOutputFolder   string         `mapstructure:"traceOutputFolder"`
+	PlacementPolicy     string         `mapstructure:"placementPolicy"`
+	Persistence         bool           `mapstructure:"persistence"`
+	Profiler            ProfilerConfig `mapstructure:"profiler"`
+	RedisConf           RedisConf      `mapstructure:"redis"`
+	Reconstruct         bool           `mapstructure:"reconstruct"`
+	LoadBalancerAddress string         `mapstructure:"loadBalancerAddress"`
+	RemoveWorkerNode    bool           `mapstructure:"removeWorkerNode"`
+	RemoveDataplane     bool           `mapstructure:"removeDataplane"`
+	PrecreateSnapshots  bool           `mapstructure:"precreateSnapshots"`
 }
 
 type DataPlaneConfig struct {