Implement structured logging

Code updated with klogs and introduce the corresponding flag for the
klog level in TAS deployment.

Author: togashidm

extender/scheduler.go

@@ -5,17 +5,18 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"io/ioutil"
-	"log"
 	"net/http"
 	"time"
+
+	"k8s.io/klog/v2"
 )
 
 //postOnly check if the method type is POST
 func postOnly(next http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodPost {
 			w.WriteHeader(http.StatusMethodNotAllowed)
-			log.Print("method Type not POST")
+			klog.V(2).InfoS("method Type not POST", "component", "extender")
 			return
 		}
 		next.ServeHTTP(w, r)
@@ -27,7 +28,7 @@ func contentLength(next http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if r.ContentLength > 1*1000*1000*1000 {
 			w.WriteHeader(http.StatusInternalServerError)
-			log.Print("request size too large")
+			klog.V(2).InfoS("request size too large", "component", "extender")
 			return
 		}
 		next.ServeHTTP(w, r)
@@ -40,7 +41,7 @@ func requestContentType(next http.HandlerFunc) http.HandlerFunc {
 		requestContentType := r.Header.Get("Content-Type")
 		if requestContentType != "application/json" {
 			w.WriteHeader(http.StatusNotFound)
-			log.Print("request content type not application/json")
+			klog.V(2).InfoS("request content type not application/json", "component", "extender")
 			return
 		}
 		next.ServeHTTP(w, r)
@@ -72,7 +73,7 @@ func handlerWithMiddleware(handle http.HandlerFunc) http.HandlerFunc {
 
 //error handler deals with requests sent to an invalid endpoint and returns a 404.
 func errorHandler(w http.ResponseWriter, r *http.Request) {
-	log.Print("unknown path")
+	klog.V(2).InfoS("Requested resource: '"+r.URL.Path+"' not found", "component", "extender")
 	w.Header().Add("Content-Type", "application/json")
 	w.WriteHeader(http.StatusNotFound)
 }
@@ -86,22 +87,23 @@ func (m Server) StartServer(port string, certFile string, keyFile string, caFile
 	mx.HandleFunc("/scheduler/filter", handlerWithMiddleware(m.Filter))
 	var err error
 	if unsafe {
-		log.Printf("Extender Listening on HTTP  %v", port)
+		klog.V(2).InfoS("Extender Listening on HTTP "+port, "component", "extender")
 		err = http.ListenAndServe(":"+port, mx)
 	} else {
-		log.Printf("Extender Now Listening on HTTPS  %v", port)
 		srv := configureSecureServer(port, caFile)
 		srv.Handler = mx
-		log.Fatal(srv.ListenAndServeTLS(certFile, keyFile))
+		klog.V(2).InfoS("Extender Listening on HTTPS "+port, "component", "extender")
+
+		klog.Fatal(srv.ListenAndServeTLS(certFile, keyFile))
 	}
-	log.Printf("Scheduler extender failed %v ", err)
+	klog.V(2).InfoS("Scheduler extender server failed to start "+err.Error(), "component", "extender")
 }
 
 //Configuration values including algorithms etc for the TAS scheduling endpoint.
 func configureSecureServer(port string, caFile string) *http.Server {
 	caCert, err := ioutil.ReadFile(caFile)
 	if err != nil {
-		log.Fatal(err)
+		klog.V(2).InfoS("caCert read failed: "+err.Error(), "component", "extender")
 	}
 	caCertPool := x509.NewCertPool()
 	caCertPool.AppendCertsFromPEM(caCert)

go.mod

@@ -6,5 +6,6 @@ require (
 	k8s.io/api v0.20.2
 	k8s.io/apimachinery v0.20.2
 	k8s.io/client-go v0.20.2
+	k8s.io/klog/v2 v2.4.0
 	k8s.io/metrics v0.20.2
 )

telemetry-aware-scheduling/cmd/main.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"flag"
-	"log"
 	"os"
 
 	"os/signal"
@@ -22,6 +21,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/klog/v2"
 
 	"context"
 
@@ -31,6 +31,7 @@ import (
 func main() {
 	var kubeConfig, port, certFile, keyFile, caFile, syncPeriod string
 	var unsafe bool
+	klog.InitFlags(nil)
 	flag.StringVar(&kubeConfig, "kubeConfig", "/root/.kube/config", "location of kubernetes config file")
 	flag.StringVar(&port, "port", "9001", "port on which the scheduler extender will listen")
 	flag.StringVar(&certFile, "cert", "/etc/kubernetes/pki/ca.crt", "cert file extender will use for authentication")
@@ -44,22 +45,26 @@ func main() {
 	sch := extender.Server{Scheduler: tscheduler}
 	go sch.StartServer(port, certFile, keyFile, caFile, unsafe)
 	tasController(kubeConfig, syncPeriod, cache)
+	klog.Flush()
 }
 
 //tasController The controller load the TAS policy/strategies and places them into a local cache that is available
 //to all TAS components. It also monitors the current state of policies.
 func tasController(kubeConfig string, syncPeriod string, cache *tascache.AutoUpdatingCache) {
 	kubeClient, clientConfig, err := getkubeClient(kubeConfig)
 	if err != nil {
+		klog.V(2).InfoS("Issue in getting client config: "+err.Error(), "component", "controller")
 		panic(err)
 	}
 	syncDuration, err := time.ParseDuration(syncPeriod)
 	if err != nil {
+		klog.V(2).InfoS("Sync problems in Parsing: "+err.Error(), "component", "controller")
 		panic(err)
 	}
 	metricsClient := metrics.NewClient(clientConfig)
 	telpolicyClient, _, err := telemetrypolicyclient.NewRest(*clientConfig)
 	if err != nil {
+		klog.V(2).InfoS("Rest client access to telemetrypolicy CRD problem: "+err.Error(), "component", "controller")
 		panic(err)
 	}
 	metricTicker := time.NewTicker(syncDuration)
@@ -86,7 +91,7 @@ func tasController(kubeConfig string, syncPeriod string, cache *tascache.AutoUpd
 func getkubeClient(kubeConfig string) (kubernetes.Interface, *rest.Config, error) {
 	clientConfig, err := rest.InClusterConfig()
 	if err != nil {
-		log.Print("not in cluster - trying file-based configuration")
+		klog.V(4).InfoS("not in cluster - trying file-based configuration", "component", "controller")
 		clientConfig, err = clientcmd.BuildConfigFromFlags("", kubeConfig)
 		if err != nil {
 			return nil, nil, err
@@ -102,5 +107,5 @@ func getkubeClient(kubeConfig string) (kubernetes.Interface, *rest.Config, error
 func catchInterrupt(done chan os.Signal) {
 	signal.Notify(done, syscall.SIGINT, syscall.SIGTERM)
 	<-done
-	log.Println("\nPolicy controller closed ")
+	klog.V(2).InfoS("Policy controller closed ", "component", "controller")
 }

telemetry-aware-scheduling/deploy/tas-deployment.yaml

@@ -24,6 +24,7 @@ spec:
         - --cert=/tas/cert/tls.crt
         - --key=/tas/cert/tls.key
         - --cacert=/var/run/secrets/kubernetes.io/serviceaccount/ca.crt
+        - --v=2
         image: tasextender
         imagePullPolicy: IfNotPresent
         securityContext:

telemetry-aware-scheduling/pkg/cache/autoupdating.go

@@ -3,12 +3,13 @@ package cache
 import (
 	"errors"
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
 	"github.com/intel/telemetry-aware-scheduling/telemetry-aware-scheduling/pkg/metrics"
 	telemetrypolicy "github.com/intel/telemetry-aware-scheduling/telemetry-aware-scheduling/pkg/telemetrypolicy/api/v1alpha1"
+
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -50,7 +51,7 @@ func (n *AutoUpdatingCache) updateAllMetrics(client metrics.Client) {
 		if len(name) > 0 {
 			err := n.updateMetric(client, name)
 			if err != nil {
-				log.Print(err)
+				klog.V(2).InfoS(err.Error(), "component", "controller")
 			}
 		} else {
 			delete(n.metricMap, name)
@@ -117,7 +118,7 @@ func (n *AutoUpdatingCache) WriteMetric(metricName string, data metrics.NodeMetr
 
 //DeletePolicy removes the policy removes the policy object at the given namespace/name string from the cache
 func (n *AutoUpdatingCache) DeletePolicy(namespace string, policyName string) error {
-	log.Print("deleting", fmt.Sprintf(policyPath, namespace, policyName))
+	klog.V(2).InfoS("deleting "+fmt.Sprintf(policyPath, namespace, policyName), "component", "controller")
 	n.delete(fmt.Sprintf(policyPath, namespace, policyName))
 	return nil
 }

telemetry-aware-scheduling/pkg/cache/autoupdating_test.go

@@ -1,13 +1,15 @@
 package cache
 
 import (
-	"log"
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
 
 	"github.com/intel/telemetry-aware-scheduling/telemetry-aware-scheduling/pkg/metrics"
 	telemetrypolicy "github.com/intel/telemetry-aware-scheduling/telemetry-aware-scheduling/pkg/telemetrypolicy/api/v1alpha1"
+
+	"k8s.io/klog/v2"
 )
 
 func TestNodeMetricsCache_PeriodicUpdate(t *testing.T) {
@@ -55,7 +57,8 @@ func TestNodeMetricsCache_PeriodicUpdate(t *testing.T) {
 				t.Error(err)
 			}
 			if atStart[tt.queriedNode].Value == atEnd[tt.queriedNode].Value {
-				log.Print(atStart[tt.queriedNode].Value, atEnd[tt.queriedNode].Value)
+				msg := fmt.Sprint(atStart[tt.queriedNode].Value, atEnd[tt.queriedNode].Value)
+				klog.InfoS(msg, "component", "testing")
 				t.Fail()
 			}
 		})

telemetry-aware-scheduling/pkg/controller/controller.go

@@ -5,7 +5,7 @@ package controller
 import (
 	"context"
 	"errors"
-	"log"
+	"fmt"
 
 	strategy "github.com/intel/telemetry-aware-scheduling/telemetry-aware-scheduling/pkg/strategies/core"
 	"github.com/intel/telemetry-aware-scheduling/telemetry-aware-scheduling/pkg/strategies/deschedule"
@@ -15,13 +15,15 @@ import (
 	core "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
 )
 
 //Run starts the controller watching on the Informer queue and doesnt' stop it until the Done signal is received from context
 func (controller *TelemetryPolicyController) Run(context context.Context) {
-	log.Print("Watching Telemetry Policies ")
+	klog.V(2).InfoS("Watching Telemetry Policies", "component", "controller")
 	_, err := controller.watch(context)
 	if err != nil {
+		klog.V(2).InfoS(err.Error(), "component", "controller")
 		panic(err)
 	}
 	<-context.Done()
@@ -54,20 +56,20 @@ func (controller *TelemetryPolicyController) watch(context context.Context) (cac
 func (controller *TelemetryPolicyController) onAdd(obj interface{}) {
 	pol, ok := obj.(*telemetrypolicy.TASPolicy)
 	if !ok {
-		log.Print("cannot add policy: not recognized as a telemetry policy")
+		klog.V(4).InfoS("cannot add policy: not recognized as a telemetry policy", "component", "controller")
 		return
 	}
 	polCopy := pol.DeepCopy()
 	err := controller.WritePolicy(polCopy.Namespace, polCopy.Name, *polCopy)
 	if err != nil {
-		log.Print("policy not added to cache: ", err)
+		klog.V(2).InfoS("Policy not added to cache: "+err.Error(), "component", "controller")
 		return
 	}
 	for _, name := range controller.Enforcer.RegisteredStrategyTypes() {
-		log.Printf("registering %v from %v", name, pol.Name)
+		klog.V(4).InfoS("registering "+name+" from "+pol.Name, "component", "controller")
 		strt, err := castStrategy(name, polCopy.Spec.Strategies[name])
 		if err != nil {
-			log.Print(err)
+			klog.V(2).InfoS(err.Error(), "component", "controller")
 			return
 		}
 		strt.SetPolicyName(polCopy.ObjectMeta.Name)
@@ -76,11 +78,11 @@ func (controller *TelemetryPolicyController) onAdd(obj interface{}) {
 		for _, rule := range ruleset[name].Rules {
 			err := controller.WriteMetric(rule.Metricname, nil)
 			if err == nil {
-				log.Print("Added " + rule.Metricname)
+				klog.V(2).InfoS("Added "+rule.Metricname, "component", "controller")
 			}
 		}
 	}
-	log.Println("Added policy,", polCopy.Name)
+	klog.V(2).InfoS("Added policy, "+polCopy.Name, "component", "controller")
 }
 
 //castStrategy takes in a TASpolicy and returns its specific type based on the structure of the policy file.
@@ -107,34 +109,35 @@ func (controller *TelemetryPolicyController) onUpdate(old, new interface{}) {
 	polCopy := newPol.DeepCopy()
 	err := controller.WritePolicy(polCopy.Namespace, polCopy.Name, *polCopy)
 	if err != nil {
-		log.Print("cached policy not updated: ", err)
+		msg := fmt.Sprintf("cached policy not updated %v", err)
+		klog.V(2).InfoS(msg, "component", "controller")
 		return
 	}
-	log.Println("Policy: ", polCopy.Name, " updated")
+	klog.V(2).InfoS("Policy: "+polCopy.Name+" updated", "component", "controller")
 	for _, name := range controller.Enforcer.RegisteredStrategyTypes() {
 		oldStrat, err := castStrategy(name, oldPol.Spec.Strategies[name])
 		if err != nil {
-			log.Print(err)
+			klog.V(2).InfoS(err.Error(), "component", "controller")
 			return
 		}
 		controller.Enforcer.RemoveStrategy(oldStrat, oldStrat.StrategyType())
 		for _, rule := range oldPol.Spec.Strategies[oldStrat.StrategyType()].Rules {
 			err := controller.DeleteMetric(rule.Metricname)
 			if err != nil {
-				log.Print(err)
+				klog.V(2).InfoS(err.Error(), "component", "controller")
 			}
 		}
 		strt, err := castStrategy(name, polCopy.Spec.Strategies[name])
 		if err != nil {
-			log.Print(err)
+			klog.V(2).InfoS(err.Error(), "component", "controller")
 			return
 		}
 		strt.SetPolicyName(polCopy.ObjectMeta.Name)
 		controller.Enforcer.AddStrategy(strt, name)
 		for _, rule := range polCopy.Spec.Strategies[name].Rules {
 			err := controller.WriteMetric(rule.Metricname, nil)
 			if err != nil {
-				log.Print(err)
+				klog.V(2).InfoS(err.Error(), "component", "controller")
 			}
 		}
 	}
@@ -147,22 +150,22 @@ func (controller *TelemetryPolicyController) onDelete(obj interface{}) {
 	for _, name := range controller.Enforcer.RegisteredStrategyTypes() {
 		strt, err := castStrategy(name, polCopy.Spec.Strategies[name])
 		if err != nil {
-			log.Print(err)
+			klog.V(2).InfoS(err.Error(), "component", "controller")
 			return
 		}
 		strt.SetPolicyName(pol.Name)
 		controller.Enforcer.RemoveStrategy(strt, strt.StrategyType())
 		for _, rule := range polCopy.Spec.Strategies[strt.StrategyType()].Rules {
 			err := controller.DeleteMetric(rule.Metricname)
 			if err != nil {
-				log.Print(err)
+				klog.V(2).InfoS(err.Error(), "component", "controller")
 			}
 		}
 	}
 	err := controller.DeletePolicy(polCopy.Namespace, polCopy.Name)
 	if err != nil {
-		log.Print(err)
+		klog.V(4).InfoS(err.Error(), "component", "controller")
 		return
 	}
-	log.Println("Policy: ", polCopy.Name, " deleted")
+	klog.V(2).InfoS("Policy: "+polCopy.Name+" deleted", "component", "controller")
 }

telemetry-aware-scheduling/pkg/metrics/client_test.go

@@ -4,11 +4,12 @@ import (
 	"bytes"
 	"fmt"
 	"io/ioutil"
-	"log"
 	"reflect"
 	"testing"
 	"time"
 
+	"k8s.io/klog/v2"
+
 	"k8s.io/client-go/tools/clientcmd"
 
 	v1 "k8s.io/api/core/v1"
@@ -30,7 +31,8 @@ var baseTimeStamp = time.Date(2019, time.May, 20, 12, 25, 00, 0, time.UTC)
 func dummyRestClientConfig() *restclient.Config {
 	tmpFile, err := ioutil.TempFile("", "cmdtests_temp")
 	if err != nil {
-		panic(fmt.Sprintf("unable to create a fake client config: %v", err))
+		klog.InfoS("Unable to create a fake client config: "+err.Error(), "component", "testing")
+		panic(err)
 	}
 	loadingRules := &clientcmd.ClientConfigLoadingRules{
 		Precedence:     []string{tmpFile.Name()},
@@ -41,7 +43,7 @@ func dummyRestClientConfig() *restclient.Config {
 	clientConfig := clientcmd.NewInteractiveDeferredLoadingClientConfig(loadingRules, overrides, fallbackReader)
 	restConfig, err := clientConfig.ClientConfig()
 	if err != nil {
-		log.Fatalf("Can't create dummy rest client config %v ", err)
+		klog.InfoS("Can't create dummy rest client config: "+err.Error(), "component", "testing")
 	}
 	return restConfig
 }
@@ -135,7 +137,7 @@ func TestNewClient(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			got := NewClient(tt.args.config)
 			if reflect.TypeOf(got) != reflect.TypeOf(dummyRestClientConfig()) {
-				log.Print("No real test implemented here")
+				klog.InfoS("No real test implemented here", "component", "testing")
 				//TODO:add some better verification constructor has worked here.
 			}
 		})