feat: added kosmos integration (#13)

* feat: added kosmos integration

* fix: constant values fro http-metrics sidecar and better resource checks

* fix:  update dispatcher service port

* fix: improve diff performance

* refactor: remove comments

.github/workflows/base-pipeline.yml

@@ -17,10 +17,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
 
-    - name: Set up Go 1.15
+    - name: Set up Go 1.16
       uses: actions/setup-go@v2
       with:
-        go-version: ^1.15
+        go-version: ^1.16
 
     - name: Check out code into the Go module directory
       uses: actions/checkout@v2

.github/workflows/e2e.yaml

@@ -21,7 +21,7 @@ jobs:
       - name: Set up Go 1.15
         uses: actions/setup-go@v2
         with:
-          go-version: ^1.15
+          go-version: 1.15
 
 #      - name: Cache Go modules
 #        uses: actions/cache@v2
@@ -34,11 +34,15 @@ jobs:
       - name: Get dependencies
         run: |
           go get -v -t -d ./...
+          
+      - name: Get vendors
+        run: |
+          go mod vendor
 
       - uses: actions/checkout@v2
 
       - name: Create k8s Kind Cluster
-        uses: helm/kind-action@v1.1.0
+        uses: helm/kind-action@v1.2.0
         with:
           node_image: "systemautoscaler/kindest-node:latest"
           config: "./config/cluster-conf/kind.conf"

config/cluster-conf/kind.conf

@@ -11,7 +11,3 @@ nodes:
 - role: worker
 - role: worker
 - role: worker
-- role: worker
-- role: worker
-- role: worker
-- role: worker

config/deploy/dispatcher.yaml

@@ -44,7 +44,8 @@ metadata:
     app: dispatcher
 spec:
   ports:
-    - port: 8080
+    - port: 80
+      targetPort: 8080
   selector:
     app: dispatcher
 

deploy/config/deploy/dispatcher.yaml

@@ -37,3 +37,16 @@ spec:
             valueFrom:
               fieldRef:
                 fieldPath: spec.nodeName
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: dispatcher
+  labels:
+    app: dispatcher
+spec:
+  ports:
+    - port: 80
+      targetPort: 8080
+  selector:
+    app: dispatcher

go.mod

@@ -14,15 +14,17 @@ require (
 	github.com/onsi/gomega v1.14.0
 	github.com/openfaas/faas-netes v0.0.0-20210722152123-f0519195a78b
 	github.com/stretchr/testify v1.7.0
+	golang.org/x/oauth2 v0.0.0-20210402161424-2e8d93401602 // indirect
+	golang.org/x/tools v0.1.5 // indirect
 	k8s.io/api v0.21.3
+	k8s.io/apiextensions-apiserver v0.21.3 // indirect
 	k8s.io/apimachinery v0.21.3
 	k8s.io/client-go v0.21.3
-	k8s.io/code-generator v0.22.2
+	k8s.io/code-generator v0.21.3
 	k8s.io/klog/v2 v2.9.0
 	k8s.io/kube-openapi v0.0.0-20210527164424-3c818078ee3d
 	k8s.io/metrics v0.21.3
 	k8s.io/utils v0.0.0-20210707171843-4b05e18ac7d9
 	sigs.k8s.io/controller-runtime v0.9.3
-	sigs.k8s.io/controller-tools v0.6.2 // indirect
 
 )

hack/tools.go

@@ -1,6 +1,9 @@
+//go:build tools
+// +build tools
+
 package tools
 
 import (
-	_ "k8s.io/code-generator" // This package imports things required by build scripts, to force `go mod` to see them as dependencies
+	_ "k8s.io/code-generator" // This package imports things required by build scripts, to force go mod to see them as dependencies
 	_ "k8s.io/kube-openapi/cmd/openapi-gen"
 )

pkg/community-controller/pkg/controller/sync.go

@@ -3,7 +3,6 @@ package controller
 import (
 	"context"
 	"fmt"
-
 	"github.com/lterrac/edge-autoscaler/pkg/apis/edgeautoscaler/v1alpha1"
 	openfaasv1 "github.com/openfaas/faas-netes/pkg/apis/openfaas/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -20,6 +19,12 @@ import (
 	"k8s.io/klog/v2"
 )
 
+const (
+	HttpMetricsPort = 8080
+	HttpMetricsCpu  = 150
+	HttpMetricsMemory = 250000000
+)
+
 func (c *CommunityController) runScheduler(_ string) error {
 
 	klog.Infof("Rescheduling community %s/%s", c.communityNamespace, c.communityName)
@@ -215,6 +220,7 @@ func newPod(function *openfaasv1.Function, cs *v1alpha1.CommunitySchedule, node
 				ealabels.FunctionNameLabel:                                   function.Name,
 				ealabels.CommunityLabel.WithNamespace(cs.Namespace).String(): cs.Name,
 				ealabels.NodeLabel:                                           node,
+				"autoscaling": "vertical",
 			},
 			OwnerReferences: []metav1.OwnerReference{
 				*metav1.NewControllerRef(function, schema.GroupVersionKind{
@@ -231,15 +237,51 @@ func newPod(function *openfaasv1.Function, cs *v1alpha1.CommunitySchedule, node
 					Name:  function.Spec.Name,
 					Image: function.Spec.Image,
 					Ports: []corev1.ContainerPort{
-						{ContainerPort: int32(8080), Protocol: corev1.ProtocolTCP},
+						{ContainerPort: int32(HttpMetricsPort), Protocol: corev1.ProtocolTCP},
 					},
-					//ImagePullPolicy: corev1.PullPolicy(factory.Factory.Config.ImagePullPolicy),
+					ImagePullPolicy: corev1.PullAlways,
 					Env:       envVars,
 					Resources: *resources,
 					// TODO: add probe with Function Factory
 					//LivenessProbe:   probes.Liveness,
 					//ReadinessProbe:  probes.Readiness,
 				},
+				{
+					Name:  "http-metrics",
+					Image: "systemautoscaler/http-metrics:0.1.0",
+					Ports: []corev1.ContainerPort{
+						{ContainerPort: int32(8000), Protocol: corev1.ProtocolTCP},
+					},
+					ImagePullPolicy: corev1.PullAlways,
+					Env: []corev1.EnvVar{
+						{
+							Name:  "ADDRESS",
+							Value: "localhost",
+						},
+						{
+							Name:  "PORT",
+							Value: "8080",
+						},
+						{
+							Name:  "WINDOW_SIZE",
+							Value: "30s",
+						},
+						{
+							Name:  "WINDOW_GRANULARITY",
+							Value: "1ms",
+						},
+					},
+					Resources: corev1.ResourceRequirements{
+						Limits: map[corev1.ResourceName]resource.Quantity{
+							corev1.ResourceCPU: *resource.NewMilliQuantity(HttpMetricsCpu, resource.BinarySI),
+							corev1.ResourceMemory: *resource.NewQuantity(HttpMetricsMemory, resource.BinarySI),
+						},
+						Requests: map[corev1.ResourceName]resource.Quantity{
+							corev1.ResourceCPU: *resource.NewMilliQuantity(HttpMetricsCpu, resource.BinarySI),
+							corev1.ResourceMemory: *resource.NewQuantity(HttpMetricsMemory, resource.BinarySI),
+						},
+					},
+				},
 			},
 		},
 	}
@@ -277,35 +319,43 @@ func makeResources(function *openfaasv1.Function) (*corev1.ResourceRequirements,
 	}
 
 	// Set Memory limits
-	if function.Spec.Limits != nil && len(function.Spec.Limits.Memory) > 0 {
-		qty, err := resource.ParseQuantity(function.Spec.Limits.Memory)
+	if function.Spec.Limits != nil && len(function.Spec.Limits.Memory) > 0 &&
+		function.Spec.Requests != nil && len(function.Spec.Requests.Memory) > 0 {
+		limit, err := resource.ParseQuantity(function.Spec.Limits.Memory)
 		if err != nil {
 			return resources, err
 		}
-		resources.Limits[corev1.ResourceMemory] = qty
-	}
-	if function.Spec.Requests != nil && len(function.Spec.Requests.Memory) > 0 {
-		qty, err := resource.ParseQuantity(function.Spec.Requests.Memory)
+		request, err := resource.ParseQuantity(function.Spec.Requests.Memory)
 		if err != nil {
 			return resources, err
 		}
-		resources.Requests[corev1.ResourceMemory] = qty
+		if limit.MilliValue() != request.MilliValue() {
+			return resources, fmt.Errorf("function %s/%s must have same memory resource requests and limits", function.Namespace, function.Name)
+		}
+		resources.Requests[corev1.ResourceMemory] = request
+		resources.Limits[corev1.ResourceMemory] = limit
+	} else {
+		return resources, fmt.Errorf("function %s/%s must have memory resource requests and limits", function.Namespace, function.Name)
 	}
 
 	// Set CPU limits
-	if function.Spec.Limits != nil && len(function.Spec.Limits.CPU) > 0 {
-		qty, err := resource.ParseQuantity(function.Spec.Limits.CPU)
+	if function.Spec.Limits != nil && len(function.Spec.Limits.CPU) > 0 &&
+		function.Spec.Requests != nil && len(function.Spec.Requests.CPU) > 0 {
+		limit, err := resource.ParseQuantity(function.Spec.Limits.CPU)
 		if err != nil {
 			return resources, err
 		}
-		resources.Limits[corev1.ResourceCPU] = qty
-	}
-	if function.Spec.Requests != nil && len(function.Spec.Requests.CPU) > 0 {
-		qty, err := resource.ParseQuantity(function.Spec.Requests.CPU)
+		request, err := resource.ParseQuantity(function.Spec.Requests.CPU)
 		if err != nil {
 			return resources, err
 		}
-		resources.Requests[corev1.ResourceCPU] = qty
+		if limit.MilliValue() != request.MilliValue() {
+			return resources, fmt.Errorf("function %s/%s must have same cpu resource requests and limits", function.Namespace, function.Name)
+		}
+		resources.Requests[corev1.ResourceCPU] = request
+		resources.Limits[corev1.ResourceCPU] = limit
+	} else {
+		return resources, fmt.Errorf("function %s/%s must have cpu resource requests and limits", function.Namespace, function.Name)
 	}
 
 	return resources, nil

pkg/dispatcher/pkg/balancer/load-balancer.go

@@ -111,8 +111,10 @@ func (lb *LoadBalancer) Balance(w http.ResponseWriter, r *http.Request) {
 		requestData.StatusCode = res.StatusCode
 	}
 
+
 	if resErr != nil {
-		http.Error(w, "Bad gateway", http.StatusBadGateway)
+		klog.Errorf("Bad gateway, error: %v", resErr.Error())
+		http.Error(w, fmt.Sprintf("Bad gateway, error: %v", resErr.Error()), http.StatusBadGateway)
 		return
 	}
 

pkg/dispatcher/pkg/controller/sync.go

@@ -132,7 +132,7 @@ func (c *LoadBalancerController) syncCommunitySchedule(key string) error {
 						continue
 					}
 
-					destinationURL, err := url.Parse(fmt.Sprintf("http://%s:%d", pod.Status.PodIP, pod.Spec.Containers[0].Ports[0].ContainerPort))
+					destinationURL, err := url.Parse(fmt.Sprintf("http://%s:%d", pod.Status.PodIP, 8000))
 
 					if err != nil {
 						utilruntime.HandleError(fmt.Errorf("error parsing function url: %s", err))