CNS to start telemetry process and connect to it

cni/network/plugin/main.go

@@ -135,30 +135,6 @@ func handleIfCniUpdate(update func(*skel.CmdArgs) error) (bool, error) {
 	return isupdate, nil
 }
 
-func connectToTelemetryService(tb *telemetry.TelemetryBuffer) {
-	path := fmt.Sprintf("%v/%v", telemetry.CniInstallDir, telemetry.TelemetryServiceProcessName)
-	args := []string{"-d", telemetry.CniInstallDir}
-
-	for attempt := 0; attempt < 2; attempt++ {
-		if err := tb.Connect(); err != nil {
-			log.Printf("Connection to telemetry socket failed: %v", err)
-			tb.Cleanup(telemetry.FdName)
-
-			if isExists, _ := common.CheckIfFileExists(path); !isExists {
-				log.Printf("Skip starting telemetry service as file didn't exist")
-				return
-			}
-
-			telemetry.StartTelemetryService(path, args)
-			telemetry.WaitForTelemetrySocket(telemetryNumRetries, telemetryWaitTimeInMilliseconds)
-		} else {
-			tb.Connected = true
-			log.Printf("Connected to telemetry service")
-			return
-		}
-	}
-}
-
 // Main is the entry point for CNI network plugin.
 func main() {
 
@@ -196,7 +172,7 @@ func main() {
 	}
 
 	tb := telemetry.NewTelemetryBuffer("")
-	connectToTelemetryService(tb)
+	tb.ConnectToTelemetryService(telemetryNumRetries, telemetryWaitTimeInMilliseconds)
 	defer tb.Close()
 
 	t := time.Now()

cns/service/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/platform"
 	"github.com/Azure/azure-container-networking/store"
+	"github.com/Azure/azure-container-networking/telemetry"
 )
 
 const (
@@ -121,13 +122,6 @@ var args = acn.ArgumentList{
 		Type:         "bool",
 		DefaultValue: false,
 	},
-	{
-		Name:         acn.OptReportToHostInterval,
-		Shorthand:    acn.OptReportToHostIntervalAlias,
-		Description:  "Set interval in ms to report to host",
-		Type:         "int",
-		DefaultValue: "60000",
-	},
 	{
 		Name:         acn.OptCNIPath,
 		Shorthand:    acn.OptCNIPathAlias,
@@ -142,6 +136,13 @@ var args = acn.ArgumentList{
 		Type:         "string",
 		DefaultValue: platform.K8SNetConfigPath + string(os.PathSeparator) + defaultCNINetworkConfigFileName,
 	},
+	{
+		Name:         acn.OptTelemetry,
+		Shorthand:    acn.OptTelemetryAlias,
+		Description:  "Set to false to disable telemetry",
+		Type:         "bool",
+		DefaultValue: true,
+	},
 }
 
 // Prints description and version information.
@@ -168,7 +169,7 @@ func main() {
 	ipamQueryInterval, _ := acn.GetArg(acn.OptIpamQueryInterval).(int)
 	stopcnm = acn.GetArg(acn.OptStopAzureVnet).(bool)
 	vers := acn.GetArg(acn.OptVersion).(bool)
-	// reportToHostInterval := acn.GetArg(acn.OptReportToHostInterval).(int)
+	telemetryEnabled := acn.GetArg(acn.OptTelemetry).(bool)
 
 	if vers {
 		printVersion()
@@ -197,7 +198,8 @@ func main() {
 		return
 	}
 
-	if logger := log.GetStd(); logger != nil {
+	// Set-up channel for CNS telemetry if it's enabled (enabled by default)
+	if logger := log.GetStd(); logger != nil && telemetryEnabled {
 		logger.SetChannel(reports)
 	}
 
@@ -231,10 +233,13 @@ func main() {
 
 	// Start CNS.
 	if httpRestService != nil {
-		// go telemetry.SendCnsTelemetry(reportToHostInterval,
-		// 	reports,
-		// 	httpRestService.(*restserver.HTTPRestService),
-		// 	telemetryStopProcessing)
+		if telemetryEnabled {
+			go telemetry.SendCnsTelemetry(
+				reports,
+				httpRestService.(*restserver.HTTPRestService),
+				telemetryStopProcessing)
+		}
+
 		err = httpRestService.Start(&config)
 		if err != nil {
 			log.Errorf("Failed to start CNS, err:%v.\n", err)

common/config.go

@@ -71,4 +71,8 @@ const (
 	// Telemetry config Location
 	OptTelemetryConfigDir      = "telemetry-config-file"
 	OptTelemetryConfigDirAlias = "d"
+
+	// Disable Telemetry
+	OptTelemetry      = "telemetry"
+	OptTelemetryAlias = "dt"
 )

log/logger.go

@@ -216,7 +216,9 @@ func (logger *Logger) Debugf(format string, args ...interface{}) {
 // Errorf logs a formatted string at info level and sends the string to TelemetryBuffer.
 func (logger *Logger) Errorf(format string, args ...interface{}) {
 	logger.Printf(format, args...)
-	// go func() {
-	// 	logger.reports <- fmt.Sprintf(format, args...)
-	// }()
+	go func() {
+		if logger.reports != nil {
+			logger.reports <- fmt.Sprintf(format, args...)
+		}
+	}()
 }

telemetry/cnstelemetry.go

@@ -16,27 +16,21 @@ import (
 
 const (
 	// CNSTelemetryFile - telemetry file path.
-	CNSTelemetryFile           = platform.CNSRuntimePath + "AzureCNSTelemetry.json"
-	errorcodePrefix            = 5
-	heartbeatIntervalInMinutes = 30
-	retryWaitTimeInSeconds     = 60
+	CNSTelemetryFile                = platform.CNSRuntimePath + "AzureCNSTelemetry.json"
+	errorcodePrefix                 = 5
+	heartbeatIntervalInMinutes      = 30
+	retryWaitTimeInSeconds          = 60
+	telemetryNumRetries             = 5
+	telemetryWaitTimeInMilliseconds = 200
 )
 
 // SendCnsTelemetry - handles cns telemetry reports
-func SendCnsTelemetry(interval int, reports chan interface{}, service *restserver.HTTPRestService, telemetryStopProcessing chan bool) {
+func SendCnsTelemetry(reports chan interface{}, service *restserver.HTTPRestService, telemetryStopProcessing chan bool) {
 
 CONNECT:
-	telemetryBuffer := NewTelemetryBuffer("")
-	err := telemetryBuffer.StartServer()
-	if err == nil || telemetryBuffer.FdExists {
-		if err := telemetryBuffer.Connect(); err != nil {
-			log.Printf("[CNS-Telemetry] Failed to establish telemetry manager connection.")
-			time.Sleep(time.Second * retryWaitTimeInSeconds)
-			goto CONNECT
-		}
-
-		go telemetryBuffer.BufferAndPushData(time.Duration(0))
-
+	tb := NewTelemetryBuffer("")
+	tb.ConnectToTelemetryService(telemetryNumRetries, telemetryWaitTimeInMilliseconds)
+	if tb.Connected {
 		heartbeat := time.NewTicker(time.Minute * heartbeatIntervalInMinutes).C
 		reportMgr := ReportManager{
 			ContentType: ContentType,
@@ -63,7 +57,7 @@ CONNECT:
 
 				reflect.ValueOf(reportMgr.Report).Elem().FieldByName("EventMessage").SetString(msg.(string))
 			case <-telemetryStopProcessing:
-				telemetryBuffer.Cancel()
+				tb.Close()
 				return
 			}
 
@@ -79,16 +73,12 @@ CONNECT:
 			report, err := reportMgr.ReportToBytes()
 			if err == nil {
 				// If write fails, try to re-establish connections as server/client
-				if _, err = telemetryBuffer.Write(report); err != nil {
+				if _, err = tb.Write(report); err != nil {
 					log.Printf("[CNS-Telemetry] Telemetry write failed: %v", err)
-					telemetryBuffer.Cancel()
+					tb.Close()
 					goto CONNECT
 				}
 			}
 		}
-	} else {
-		log.Printf("[CNS-Telemetry] Failed to start telemetry manager server.")
-		time.Sleep(time.Second * retryWaitTimeInSeconds)
-		goto CONNECT
 	}
 }

telemetry/telemetry.go

@@ -26,9 +26,9 @@ const (
 	NPMTelemetryFile = platform.NPMRuntimePath + "AzureNPMTelemetry.json"
 	// CNITelemetryFile Path.
 	CNITelemetryFile = platform.CNIRuntimePath + "AzureCNITelemetry.json"
-
-	metadataURL = "http://169.254.169.254/metadata/instance?api-version=2017-08-01&format=json"
+	// ContentType of JSON
 	ContentType = "application/json"
+	metadataURL = "http://169.254.169.254/metadata/instance?api-version=2017-08-01&format=json"
 )
 
 // OS Details structure.

telemetry/telemetrybuffer.go

@@ -11,6 +11,8 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -464,3 +466,39 @@ func ReadConfigFile(filePath string) (TelemetryConfig, error) {
 
 	return config, err
 }
+
+// ConnectToTelemetryService - Attempt to spawn telemetry process if it's not already running.
+func (tb *TelemetryBuffer) ConnectToTelemetryService(telemetryNumRetries, telemetryWaitTimeInMilliseconds int) {
+	path, dir := getTelemetryServiceDirectory()
+	args := []string{"-d", dir}
+	for attempt := 0; attempt < 2; attempt++ {
+		if err := tb.Connect(); err != nil {
+			log.Printf("Connection to telemetry socket failed: %v", err)
+			tb.Cleanup(FdName)
+			StartTelemetryService(path, args)
+			WaitForTelemetrySocket(telemetryNumRetries, time.Duration(telemetryWaitTimeInMilliseconds))
+		} else {
+			tb.Connected = true
+			log.Printf("Connected to telemetry service")
+			return
+		}
+	}
+}
+
+func getTelemetryServiceDirectory() (path string, dir string) {
+	path = fmt.Sprintf("%v/%v", CniInstallDir, TelemetryServiceProcessName)
+	if exists, _ := common.CheckIfFileExists(path); !exists {
+		ex, _ := os.Executable()
+		exDir := filepath.Dir(ex)
+		path = fmt.Sprintf("%v/%v", exDir, TelemetryServiceProcessName)
+		if exists, _ = common.CheckIfFileExists(path); !exists {
+			log.Printf("Skip starting telemetry service as file didn't exist")
+			return
+		}
+		dir = exDir
+	} else {
+		dir = CniInstallDir
+	}
+
+	return
+}