x/ref/lib/aws/vxray: add support for obtaining eks/k8s data (#206)

This PR adds support for obtaining the eke/kubernetes(k8s) cluster name, container id etc. Note, this is PR is not specific to EKS and can be used for other k8s systems. It relies on the convention that the k8s cluster is configured to write a configmap with the cluster's name as one of the key/value pairs. The vxray setup allows for both the name of the configmap and the key for the cluster name to be configured.

Author: cosnicolaou

x/ref/examples/echo/bin/Dockerfile.echo

@@ -0,0 +1,4 @@
+FROM alpine
+COPY echo /bin/echo
+COPY creds/ /bin/creds/
+ENTRYPOINT ["/bin/echo"]

x/ref/examples/echo/bin/Dockerfile.echod

@@ -0,0 +1,4 @@
+FROM alpine
+COPY echod /bin/echod
+COPY creds/ /bin/creds/
+ENTRYPOINT ["/bin/echod"]

x/ref/examples/echo/bin/Dockerfile.mounttabled

@@ -0,0 +1,4 @@
+FROM alpine
+COPY mounttabled /bin/mounttabled
+COPY creds/ /bin/creds/
+ENTRYPOINT ["/bin/mounttabled"]

x/ref/examples/echo/echo/echo.go

@@ -6,6 +6,7 @@
 package main
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -18,6 +19,7 @@ import (
 	"github.com/aws/aws-xray-sdk-go/xray"
 	v23 "v.io/v23"
 	"v.io/v23/context"
+	"v.io/v23/naming"
 	"v.io/v23/vtrace"
 	"v.io/x/ref/examples/echo"
 	"v.io/x/ref/lib/aws/vxray"
@@ -50,7 +52,20 @@ func main() {
 	ctx, shutdown := v23.Init()
 	defer shutdown()
 
-	ctx, _ = vxray.InitXRay(ctx, v23.GetRuntimeFlags().VtraceFlags, xray.Config{ServiceVersion: ""}, vxray.EC2Plugin(), vxray.MergeLogging(true))
+	ctx, _ = vxray.InitXRay(ctx,
+		v23.GetRuntimeFlags().VtraceFlags,
+		xray.Config{ServiceVersion: ""},
+		vxray.EC2Plugin(),
+		vxray.EKSCluster(),
+		vxray.ContainerIDAndHost(),
+		vxray.MergeLogging(true))
+
+	servers := strings.Split(serverFlag, ",")
+	if len(servers) > 0 {
+		ctx.Infof("waiting for: %v servers: %v", len(servers), servers)
+		waitForServers(ctx, servers)
+		ctx.Infof("servers ready: %v", servers)
+	}
 
 	client := echo.EchoServiceClient(nameFlag)
 
@@ -59,7 +74,7 @@ func main() {
 	if len(httpAddr) > 0 {
 		wg.Add(1)
 		go func() {
-			runHttpServer(ctx, httpAddr, client)
+			runHTTPServer(ctx, httpAddr, client)
 			wg.Done()
 		}()
 	}
@@ -73,7 +88,6 @@ func main() {
 		close(done)
 	}()
 
-	servers := strings.Split(serverFlag, ",")
 	samplingRequest := &vtrace.SamplingRequest{
 		Name: nameFlag,
 	}
@@ -106,7 +120,7 @@ func main() {
 	}()
 	select {
 	case <-done:
-		time.Sleep(1)
+		time.Sleep(time.Second * 2)
 	case <-signals.ShutdownOnSignals(ctx):
 	}
 }
@@ -115,7 +129,7 @@ func main() {
 // <addr>/call and <addr>/call?forward-to=<server>
 // will issue RPCs to the echo server.
 // <addr>/quit will cause the client to exit gracefully.
-func runHttpServer(ctx *context.T, addr string, client echo.EchoServiceClientStub) {
+func runHTTPServer(ctx *context.T, addr string, client echo.EchoServiceClientStub) {
 	xrayHandler := xray.Handler(
 		xray.NewFixedSegmentNamer("http.echo.client"),
 		http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -169,7 +183,7 @@ func callPing(ctx *context.T, client echo.EchoServiceClientStub, out io.Writer,
 	}
 	result, err := client.Ping(ctx, now, servers)
 	if err != nil {
-		ctx.Errorf("%v.%v failed: %v", nameFlag, "ping", err)
+		ctx.Errorf("%v.%v failed: %v", servers, "ping", err)
 	}
 	if len(result) < 100 {
 		fmt.Fprintln(out, result)
@@ -178,3 +192,27 @@ func callPing(ctx *context.T, client echo.EchoServiceClientStub, out io.Writer,
 	}
 	return err
 }
+
+func waitForServers(ctx *context.T, servers []string) {
+	var wg sync.WaitGroup
+	wg.Add(len(servers))
+	ns := v23.GetNamespace(ctx)
+	for _, server := range servers {
+		go func(server string) {
+			for {
+				_, err := ns.Resolve(ctx, server)
+				ctx.Infof("%v: %v: %v", server, err, errors.Is(err, naming.ErrNoSuchName))
+				if errors.Is(err, naming.ErrNoSuchName) {
+					time.Sleep(time.Second)
+					continue
+				}
+				if err == nil {
+					break
+				}
+				ctx.Infof("%v: %v\n", server, err)
+			}
+			wg.Done()
+		}(server)
+	}
+	wg.Wait()
+}

x/ref/examples/echo/echod/echod.go

@@ -76,7 +76,13 @@ func main() {
 	ctx, shutdown := v23.Init()
 	defer shutdown()
 
-	ctx, _ = vxray.InitXRay(ctx, v23.GetRuntimeFlags().VtraceFlags, xray.Config{ServiceVersion: ""}, vxray.EC2Plugin(), vxray.MergeLogging(true))
+	ctx, _ = vxray.InitXRay(ctx,
+		v23.GetRuntimeFlags().VtraceFlags,
+		xray.Config{ServiceVersion: ""},
+		vxray.EC2Plugin(),
+		vxray.EKSCluster(),
+		vxray.ContainerIDAndHost(),
+		vxray.MergeLogging(true))
 
 	ctx, server, err := v23.WithNewServer(ctx, nameFlag, echo.EchoServiceServer(&echod{}), securityflag.NewAuthorizerOrDie(ctx))
 	if err != nil {

x/ref/lib/aws/vxray/config.go

@@ -9,6 +9,7 @@ package vxray
 
 import (
 	"fmt"
+	"os"
 
 	"github.com/aws/aws-xray-sdk-go/awsplugins/beanstalk"
 	"github.com/aws/aws-xray-sdk-go/awsplugins/ec2"
@@ -18,15 +19,18 @@ import (
 	"v.io/v23/context"
 	"v.io/v23/logging"
 	"v.io/v23/vtrace"
+	"v.io/x/ref/lib/aws/vxray/internal"
 	"v.io/x/ref/lib/flags"
 	libvtrace "v.io/x/ref/lib/vtrace"
 )
 
 type options struct {
-	mergeLogging  bool
-	mapToHTTP     bool
-	newStore      bool
-	newStoreFlags flags.VtraceFlags
+	mergeLogging            bool
+	mapToHTTP               bool
+	newStore                bool
+	newStoreFlags           flags.VtraceFlags
+	configMap, configMapKey string
+	containerized           bool
 }
 
 // Option represents an option to InitXRay.
@@ -53,6 +57,44 @@ func BeanstalkPlugin() Option {
 	}
 }
 
+// KubernetesCluster configures obtaining information about the process'
+// current environment when running under Kubernetes (k8s), whether managed by
+// AWS EKS or any other control plane implementation. It requires that the
+// K8S configuration creates a configmap that contains the cluster name.
+// The configMap argument names that configmap and configMapKey
+// is the key in that configmap for the cluster name. For example, when using
+// the AWS cloudwatch/insights/xray-daemon daemonset the values for those
+// would be:
+//     /api/v1/namespaces/amazon-cloudwatch/configmaps/cluster-info
+//     cluster.name
+//
+// When configured, xray segments will contain a 'cluster_name' annotation.
+func KubernetesCluster(configMap, configMapKey string) Option {
+	return func(o *options) {
+		o.configMap, o.configMapKey = configMap, configMapKey
+	}
+}
+
+// EKSCluster calls KubernetesCluster with the values commonly used
+// with EKS clusters.
+func EKSCluster() Option {
+	return KubernetesCluster("/api/v1/namespaces/amazon-cloudwatch/configmaps/cluster-info", "cluster.name")
+}
+
+// ContainerIDAndHost requests that container id and host information be
+// obtained and added to traces. The container id is obtained by parsing
+// the /proc/self/cgroup file, and the host by call the operating system's
+// hostname function. When running under kubernetes for example, a pod's
+// name is configured as its hostname.
+//
+// When configured, xray segments will contain 'container_id' and 'container_host'
+// annotations.
+func ContainerIDAndHost() Option {
+	return func(o *options) {
+		o.containerized = true
+	}
+}
+
 // MergeLogging arrays for xray logging messages to be merged with vanadium
 // log messages.
 func MergeLogging(v bool) Option {
@@ -105,44 +147,73 @@ func (xl *xraylogger) Log(level xraylog.LogLevel, msg fmt.Stringer) {
 	}
 }
 
-func initXRay(ctx *context.T, config xray.Config, opts []Option) (*context.T, *options, error) {
-	o := &options{mapToHTTP: true}
-	for _, fn := range opts {
-		fn(o)
-	}
+func (m *manager) initXRay(ctx *context.T, config xray.Config) (*context.T, error) {
 	if err := xray.Configure(config); err != nil {
 		ctx.Errorf("failed to configure xray context: %v", err)
-		return ctx, nil, err
+		return ctx, err
 	}
-	if o.mergeLogging {
+	if m.options.mergeLogging {
 		xray.SetLogger(&xraylogger{context.LoggerFromContext(ctx)})
 	}
 	ctx, err := WithConfig(ctx, config)
-	return ctx, o, err
+	return ctx, err
 }
 
 // InitXRay configures the AWS xray service and returns a context containing
 // the xray configuration. This should only be called once. The vflags argument
 // is used solely to check if xray tracing is enabled and not to create a
-// new vtrace.Store, if a new store is required, the
+// new vtrace.Store, if a new/alternate store is required, the WithNewStore option
+// should be used to specify the store to be used.
 func InitXRay(ctx *context.T, vflags flags.VtraceFlags, config xray.Config, opts ...Option) (*context.T, error) {
 	if !vflags.EnableAWSXRay {
 		return ctx, nil
 	}
 	octx := ctx
-	ctx, options, err := initXRay(ctx, config, opts)
+	mgr := &manager{}
+	mgr.options.mapToHTTP = true
+	for _, fn := range opts {
+		fn(&mgr.options)
+	}
+	ctx, err := mgr.initXRay(ctx, config)
 	if err != nil {
 		return octx, err
 	}
-
-	if options.newStore {
-		store, err := libvtrace.NewStore(options.newStoreFlags)
+	if mgr.options.newStore {
+		store, err := libvtrace.NewStore(mgr.options.newStoreFlags)
 		if err != nil {
 			return octx, err
 		}
 		ctx = vtrace.WithStore(ctx, store)
 	}
-	mgr := &manager{mapToHTTP: options.mapToHTTP}
+	if mgr.options.containerized {
+		if hostNameErr == nil {
+			mgr.containerHost = hostName
+		} else {
+			ctx.Infof("failed to obtain host name from: %v", hostNameErr)
+		}
+		cgroupFile := "/proc/self/cgroup"
+		if cid, err := internal.GetContainerID(cgroupFile); err == nil {
+			mgr.containerID = cid
+		} else {
+			ctx.Infof("failed to obtain container id", err)
+		}
+	}
+	if cm := mgr.options.configMap; len(cm) > 0 {
+		if clusterName, err := internal.GetEKSClusterName(ctx, cm, mgr.options.configMapKey); err == nil {
+			mgr.clusterName = clusterName
+		} else {
+			ctx.Infof("failed to obtain cluster name from %v.%v: %v", cm, mgr.options.configMapKey, err)
+		}
+	}
 	ctx = vtrace.WithManager(ctx, mgr)
 	return ctx, nil
 }
+
+var (
+	hostName    string
+	hostNameErr error
+)
+
+func init() {
+	hostName, hostNameErr = os.Hostname()
+}