added application insights telemetry package to send metrics and logs (#421)

* added application insights telemetry package to send metrics and logs to appinsights

* moved a function common to both packages
acquired a lock before writing to file

* added read write lock as per chandan comment. Addressed jaeryn comments

* fixed telemetry unit test

* defined interface and added appinsights package as vendor to acn

* added vendor package.go(appinsights)

* dependencies of appinsights

* added AI dependencies

* updated unit tests

* addressed review comments

Author: tamilmani1989

Gopkg.lock

@@ -61,6 +61,10 @@
   name = "github.com/containernetworking/cni"
   revision = "fbb95fff8a5239a4295c991efa8a397d43118f7e"
 
+[[constraint]]
+  name = "github.com/microsoft/ApplicationInsights-Go"
+  revision = "d813d7725313000ad1b71627b8951323635f0572"
+
 [prune]
   go-tests = true
   unused-packages = true

Gopkg.toml

@@ -7,6 +7,7 @@ COREFILES = \
 	$(wildcard netlink/*.go) \
 	$(wildcard network/*.go) \
 	$(wildcard telemetry/*.go) \
+	$(wildcard aitelemetry/*.go) \
 	$(wildcard network/epcommon/*.go) \
 	$(wildcard network/policy/*.go) \
 	$(wildcard platform/*.go) \
@@ -313,6 +314,7 @@ test-all:
         ./netlink/ \
         ./store/ \
         ./telemetry/ \
+		./aitelemetry/ \
         ./cnm/network/ \
         ./cni/ipam/ \
         ./cns/ipamclient/ \

Makefile

@@ -0,0 +1,47 @@
+package aitelemetry
+
+import (
+	"sync"
+
+	"github.com/Azure/azure-container-networking/common"
+	"github.com/microsoft/ApplicationInsights-Go/appinsights"
+)
+
+// Application trace/log structure
+type Report struct {
+	Message          string
+	Context          string
+	CustomDimensions map[string]string
+}
+
+// Application metrics structure
+type Metric struct {
+	Name             string
+	Value            float64
+	CustomDimensions map[string]string
+}
+
+// TelmetryHandle holds appinsight handles and metadata
+type telemetryHandle struct {
+	telemetryConfig             *appinsights.TelemetryConfiguration
+	appName                     string
+	appVersion                  string
+	metadata                    common.Metadata
+	diagListener                appinsights.DiagnosticsMessageListener
+	client                      appinsights.TelemetryClient
+	enableMetadataRefreshThread bool
+	refreshTimeout              int
+	rwmutex                     sync.RWMutex
+}
+
+// Telemetry Interface to send metrics/Logs to appinsights
+type TelemetryHandle interface {
+	// TrackLog function sends report (trace) to appinsights resource. It overrides few of the existing columns with app information
+	// and for rest it uses custom dimesion
+	TrackLog(report Report)
+	// TrackMetric function sends metric to appinsights resource. It overrides few of the existing columns with app information
+	// and for rest it uses custom dimesion
+	TrackMetric(metric Metric)
+	// Close - should be called for each NewAITelemetry call. Will release resources acquired
+	Close(timeout int)
+}

aitelemetry/api.go

@@ -0,0 +1 @@
+{"location":"eastus","name":"k8s-agentpool1-42685608-0","offer":"aks","osType":"Linux","placementGroupId":"","platformFaultDomain":"0","platformUpdateDomain":"0","publisher":"microsoft-aks","resourceGroupName":"rgcnideftesttamil","sku":"aks-ubuntu-1604-201902","subscriptionId":"ea821859-912a-4d20-a4dd-e18a3ce5ba2c","tags":"aksEngineVersion:canary;creationSource:aksengine-k8s-agentpool1-42685608-0;orchestrator:Kubernetes:1.10.13;poolName:agentpool1;resourceNameSuffix:42685608","version":"2019.02.12","vmId":"6baf785b-397c-4967-9f75-cdb3d0df66c4","vmSize":"Standard_DS2_v2","KernelVersion":""}

aitelemetry/metadata_test.json

@@ -0,0 +1,176 @@
+package aitelemetry
+
+import (
+	"runtime"
+	"time"
+
+	"github.com/Azure/azure-container-networking/common"
+	"github.com/Azure/azure-container-networking/log"
+	"github.com/Azure/azure-container-networking/store"
+	"github.com/microsoft/ApplicationInsights-Go/appinsights"
+)
+
+const (
+	resourceGroupStr  = "ResourceGroup"
+	vmSizeStr         = "VMSize"
+	osVersionStr      = "OSVersion"
+	locationStr       = "Region"
+	appVersionStr     = "Appversion"
+	subscriptionIDStr = "SubscriptionID"
+	defaultTimeout    = 10
+)
+
+func messageListener() appinsights.DiagnosticsMessageListener {
+	return appinsights.NewDiagnosticsMessageListener(func(msg string) error {
+		log.Printf("[AppInsights] [%s] %s\n", time.Now().Format(time.UnixDate), msg)
+		return nil
+	})
+}
+
+func getMetadata(th *telemetryHandle) {
+	var metadata common.Metadata
+	var err error
+
+	// check if metadata in memory otherwise initiate wireserver request
+	for {
+		metadata, err = common.GetHostMetadata(metadataFile)
+		if err == nil || !th.enableMetadataRefreshThread {
+			break
+		}
+
+		log.Printf("[AppInsights] Error getting metadata %v. Sleep for %d", err, th.refreshTimeout)
+		time.Sleep(time.Duration(th.refreshTimeout) * time.Second)
+	}
+
+	//acquire write lock before writing metadata to telemetry handle
+	th.rwmutex.Lock()
+	th.metadata = metadata
+	th.rwmutex.Unlock()
+
+	// Save metadata retrieved from wireserver to a file
+	kvs, err := store.NewJsonFileStore(metadataFile)
+	if err != nil {
+		log.Printf("[AppInsights] Error initializing kvs store: %v", err)
+		return
+	}
+
+	kvs.Lock(true)
+	err = common.SaveHostMetadata(th.metadata, metadataFile)
+	kvs.Unlock(true)
+	if err != nil {
+		log.Printf("[AppInsights] saving host metadata failed with :%v", err)
+	}
+}
+
+// NewAITelemetry creates telemetry handle with user specified appinsights key.
+func NewAITelemetry(
+	key string,
+	appName string,
+	appVersion string,
+	batchSize int,
+	batchInterval int,
+	enableMetadataRefreshThread bool,
+	refreshTimeout int,
+) TelemetryHandle {
+
+	telemetryConfig := appinsights.NewTelemetryConfiguration(key)
+	telemetryConfig.MaxBatchSize = batchSize
+	telemetryConfig.MaxBatchInterval = time.Duration(batchInterval) * time.Second
+
+	th := &telemetryHandle{
+		client:                      appinsights.NewTelemetryClientFromConfig(telemetryConfig),
+		appName:                     appName,
+		appVersion:                  appVersion,
+		diagListener:                messageListener(),
+		enableMetadataRefreshThread: enableMetadataRefreshThread,
+		refreshTimeout:              refreshTimeout,
+	}
+
+	if th.enableMetadataRefreshThread {
+		go getMetadata(th)
+	} else {
+		getMetadata(th)
+	}
+
+	return th
+}
+
+// TrackLog function sends report (trace) to appinsights resource. It overrides few of the existing columns with app information
+// and for rest it uses custom dimesion
+func (th *telemetryHandle) TrackLog(report Report) {
+	// Initialize new trace message
+	trace := appinsights.NewTraceTelemetry(report.Message, appinsights.Warning)
+
+	//Override few of existing columns with metadata
+	trace.Tags.User().SetAuthUserId(runtime.GOOS)
+	trace.Tags.Operation().SetId(report.Context)
+	trace.Tags.Operation().SetParentId(th.appName)
+
+	// copy app specified custom dimension
+	for key, value := range report.CustomDimensions {
+		trace.Properties[key] = value
+	}
+
+	trace.Properties[appVersionStr] = th.appVersion
+
+	// Acquire read lock to read metadata
+	th.rwmutex.RLock()
+	metadata := th.metadata
+	th.rwmutex.RUnlock()
+
+	// Check if metadata is populated
+	if metadata.SubscriptionID != "" {
+		// copy metadata from wireserver to trace
+		trace.Tags.User().SetAccountId(th.metadata.SubscriptionID)
+		trace.Tags.User().SetId(th.metadata.VMName)
+		trace.Properties[locationStr] = th.metadata.Location
+		trace.Properties[resourceGroupStr] = th.metadata.ResourceGroupName
+		trace.Properties[vmSizeStr] = th.metadata.VMSize
+		trace.Properties[osVersionStr] = th.metadata.OSVersion
+	}
+
+	// send to appinsights resource
+	th.client.Track(trace)
+}
+
+// TrackMetric function sends metric to appinsights resource. It overrides few of the existing columns with app information
+// and for rest it uses custom dimesion
+func (th *telemetryHandle) TrackMetric(metric Metric) {
+	// Initialize new metric
+	aimetric := appinsights.NewMetricTelemetry(metric.Name, metric.Value)
+
+	// Acquire read lock to read metadata
+	th.rwmutex.RLock()
+	metadata := th.metadata
+	th.rwmutex.RUnlock()
+
+	// Check if metadata is populated
+	if metadata.SubscriptionID != "" {
+		aimetric.Properties[locationStr] = th.metadata.Location
+		aimetric.Properties[subscriptionIDStr] = th.metadata.SubscriptionID
+	}
+
+	// copy custom dimensions
+	for key, value := range metric.CustomDimensions {
+		aimetric.Properties[key] = value
+	}
+
+	// send metric to appinsights
+	th.client.Track(aimetric)
+}
+
+// Close - should be called for each NewAITelemetry call. Will release resources acquired
+func (th *telemetryHandle) Close(timeout int) {
+	if timeout <= 0 {
+		timeout = defaultTimeout
+	}
+
+	// wait for items to be sent otherwise timeout
+	<-th.client.Channel().Close(time.Duration(timeout) * time.Second)
+
+	// Remove diganostic message listener
+	if th.diagListener != nil {
+		th.diagListener.Remove()
+		th.diagListener = nil
+	}
+}

aitelemetry/telemetrywrapper.go

@@ -0,0 +1,5 @@
+package aitelemetry
+
+const (
+	metadataFile = "/tmp/azuremetadata.json"
+)

aitelemetry/telemetrywrapper_linux.go

@@ -0,0 +1,63 @@
+package aitelemetry
+
+import (
+	"os"
+	"runtime"
+	"testing"
+
+	"github.com/Azure/azure-container-networking/platform"
+)
+
+var th TelemetryHandle
+
+func TestMain(m *testing.M) {
+
+	if runtime.GOOS == "linux" {
+		platform.ExecuteCommand("cp metadata_test.json /tmp/azuremetadata.json")
+	} else {
+		platform.ExecuteCommand("copy metadata_test.json azuremetadata.json")
+	}
+
+	exitCode := m.Run()
+
+	if runtime.GOOS == "linux" {
+		platform.ExecuteCommand("rm /tmp/azuremetadata.json")
+	} else {
+		platform.ExecuteCommand("del azuremetadata.json")
+	}
+
+	os.Exit(exitCode)
+}
+
+func TestNewAITelemetry(t *testing.T) {
+	th = NewAITelemetry("00ca2a73-c8d6-4929-a0c2-cf84545ec225", "testapp", "v1.0.26", 4096, 2, false, 10)
+	if th == nil {
+		t.Errorf("Error intializing AI telemetry")
+	}
+}
+
+func TestTrackMetric(t *testing.T) {
+	metric := Metric{
+		Name:             "test",
+		Value:            1.0,
+		CustomDimensions: make(map[string]string),
+	}
+
+	metric.CustomDimensions["dim1"] = "col1"
+	th.TrackMetric(metric)
+}
+
+func TestTrackLog(t *testing.T) {
+	report := Report{
+		Message:          "test",
+		Context:          "10a",
+		CustomDimensions: make(map[string]string),
+	}
+
+	report.CustomDimensions["dim1"] = "col1"
+	th.TrackLog(report)
+}
+
+func TestClose(t *testing.T) {
+	th.Close(10)
+}

aitelemetry/telemetrywrapper_test.go

@@ -0,0 +1,10 @@
+package aitelemetry
+
+import (
+	"os"
+	"path/filepath"
+)
+
+var (
+	metadataFile = filepath.FromSlash(os.Getenv("TEMP")) + "\\azuremetadata.json"
+)