Fix NPM Regression & Remove TelemetryBuffer Sidecar (#449)

* give precedence to drop rules (over allow)

* - Moving kube-system-chain above target-sets-chain
- Add drop entry at the end of Ingress-From and Egress-To chains when there are non Allow-All* entries

* write logs to stdout (and log file) so that we can see logs via kubectl

* removing kube-system chain and fixing tests

* removing telemetry buffer

Author: jaer-tsun

npm/azure-npm.yaml

@@ -76,7 +76,7 @@ spec:
         beta.kubernetes.io/os: linux
       containers:
         - name: azure-npm
-          image: mcr.microsoft.com/containernetworking/azure-npm:v1.0.28
+          image: mcr.microsoft.com/containernetworking/azure-npm:v1.0.30
           securityContext:
             privileged: true
           env:
@@ -90,17 +90,6 @@ spec:
             mountPath: /run/xtables.lock
           - name: log
             mountPath: /var/log
-          - name: socket-dir
-            mountPath: /var/run
-          - name: tmp
-            mountPath: /tmp
-        - name: azure-vnet-telemetry
-          image: mcr.microsoft.com/containernetworking/azure-vnet-telemetry:v1.0.28
-          volumeMounts:
-          - name: socket-dir
-            mountPath: /var/run
-          - name: tmp
-            mountPath: /tmp
       hostNetwork: true
       volumes:
       - name: log
@@ -111,10 +100,4 @@ spec:
         hostPath:
           path: /run/xtables.lock
           type: File
-      - name: tmp
-        hostPath:
-          path: /tmp
-          type: Directory
-      - name: socket-dir
-        emptyDir: {}
       serviceAccountName: azure-npm

npm/iptm/iptm.go

@@ -101,32 +101,6 @@ func (iptMgr *IptablesManager) InitNpmChains() error {
 		}
 	}
 
-	// Create AZURE-NPM-KUBE-SYSTEM chain.
-	if err := iptMgr.AddChain(util.IptablesAzureKubeSystemChain); err != nil {
-		return err
-	}
-
-	// Append AZURE-NPM-KUBE-SYSTEM chain to AZURE-NPM chain.
-	entry = &IptEntry{
-		Chain: util.IptablesAzureChain,
-		Specs: []string{
-			util.IptablesJumpFlag,
-			util.IptablesAzureKubeSystemChain,
-		},
-	}
-	exists, err = iptMgr.Exists(entry)
-	if err != nil {
-		return err
-	}
-
-	if !exists {
-		iptMgr.OperationFlag = util.IptablesAppendFlag
-		if _, err = iptMgr.Run(entry); err != nil {
-			log.Errorf("Error: failed to add AZURE-NPM-KUBE-SYSTEM chain to AZURE-NPM chain.")
-			return err
-		}
-	}
-
 	// Create AZURE-NPM-INGRESS-PORT chain.
 	if err := iptMgr.AddChain(util.IptablesAzureIngressPortChain); err != nil {
 		return err

npm/npm.go

@@ -5,7 +5,6 @@ package npm
 import (
 	"fmt"
 	"os"
-	"reflect"
 	"sync"
 	"time"
 
@@ -32,9 +31,6 @@ const (
 	heartbeatIntervalInMinutes    = 30
 )
 
-// reports channel
-var reports = make(chan interface{}, 1000)
-
 // NetworkPolicyManager contains informers for pod, namespace and networkpolicy.
 type NetworkPolicyManager struct {
 	sync.Mutex
@@ -81,60 +77,6 @@ func (npMgr *NetworkPolicyManager) GetClusterState() telemetry.ClusterState {
 	return npMgr.clusterState
 }
 
-// SendNpmTelemetry updates the npm report then send it.
-func (npMgr *NetworkPolicyManager) SendNpmTelemetry() {
-	if !npMgr.TelemetryEnabled {
-		return
-	}
-
-CONNECT:
-	tb := telemetry.NewTelemetryBuffer("")
-	for {
-		tb.TryToConnectToTelemetryService()
-		if tb.Connected {
-			break
-		}
-
-		time.Sleep(time.Second * telemetryRetryTimeInSeconds)
-	}
-
-	heartbeat := time.NewTicker(time.Minute * heartbeatIntervalInMinutes).C
-	report := npMgr.reportManager.Report
-	for {
-		select {
-		case <-heartbeat:
-			clusterState := npMgr.GetClusterState()
-			v := reflect.ValueOf(report).Elem().FieldByName("ClusterState")
-			if v.CanSet() {
-				v.FieldByName("PodCount").SetInt(int64(clusterState.PodCount))
-				v.FieldByName("NsCount").SetInt(int64(clusterState.NsCount))
-				v.FieldByName("NwPolicyCount").SetInt(int64(clusterState.NwPolicyCount))
-			}
-			reflect.ValueOf(report).Elem().FieldByName("ErrorMessage").SetString("heartbeat")
-		case msg := <-reports:
-			reflect.ValueOf(report).Elem().FieldByName("ErrorMessage").SetString(msg.(string))
-			fmt.Println(msg.(string))
-		}
-
-		reflect.ValueOf(report).Elem().FieldByName("Timestamp").SetString(time.Now().UTC().String())
-		// TODO: Remove below line after the host change is rolled out
-		reflect.ValueOf(report).Elem().FieldByName("EventMessage").SetString(time.Now().UTC().String())
-
-		report, err := npMgr.reportManager.ReportToBytes()
-		if err != nil {
-			log.Logf("ReportToBytes failed: %v", err)
-			continue
-		}
-
-		// If write fails, try to re-establish connections as server/client
-		if _, err = tb.Write(report); err != nil {
-			log.Logf("Telemetry write failed: %v", err)
-			tb.Close()
-			goto CONNECT
-		}
-	}
-}
-
 // restore restores iptables from backup file
 func (npMgr *NetworkPolicyManager) restore() {
 	iptMgr := iptm.NewIptablesManager()
@@ -233,11 +175,6 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
 		TelemetryEnabled: true,
 	}
 
-	// Set-up channel for Azure-NPM telemetry if it's enabled (enabled by default)
-	if logger := log.GetStd(); logger != nil && npMgr.TelemetryEnabled {
-		logger.SetChannel(reports)
-	}
-
 	clusterID := util.GetClusterID(npMgr.nodeName)
 	clusterState := npMgr.GetClusterState()
 	npMgr.reportManager.Report.(*telemetry.NPMReport).GetReport(clusterID, npMgr.nodeName, npmVersion, serverVersion.GitVersion, clusterState)

npm/plugin/main.go

@@ -22,7 +22,7 @@ var version string
 func initLogging() error {
 	log.SetName("azure-npm")
 	log.SetLevel(log.LevelInfo)
-	if err := log.SetTarget(log.TargetLogfile); err != nil {
+	if err := log.SetTarget(log.TargetStdOutAndLogFile); err != nil {
 		log.Logf("Failed to configure logging, err:%v.", err)
 		return err
 	}
@@ -60,8 +60,6 @@ func main() {
 
 	npMgr := npm.NewNetworkPolicyManager(clientset, factory, version)
 
-	go npMgr.SendNpmTelemetry()
-
 	if err = npMgr.Start(wait.NeverStop); err != nil {
 		log.Logf("npm failed with error %v.", err)
 		panic(err.Error)