x/ref/lib/vxray: add support for logging eks cluster, container info

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)
 
@@ -73,7 +88,6 @@ func main() {
 		close(done)
 	}()
 
-	servers := strings.Split(serverFlag, ",")
 	samplingRequest := &vtrace.SamplingRequest{
 		Name: nameFlag,
 	}
@@ -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,43 @@ 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, the
+//
+// 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 +146,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()
+}

x/ref/lib/aws/vxray/internal/eks_info.go

@@ -0,0 +1,111 @@
+package internal
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+)
+
+const (
+	k8sServiceAccountPrefix = "/var/run/secrets/kubernetes.io/serviceaccount"
+	k8sCert                 = "ca.crt"
+	k8sToken                = "token"
+	k8sAPIHost              = "kubernetes.default.svc"
+	kContainerIDLen         = 64
+)
+
+func GetEKSClusterName(ctx context.Context, configMap, keyName string) (string, error) {
+	cm, err := getConfigMap(ctx, configMap)
+	if err != nil {
+		return "", err
+	}
+	name, ok := cm[keyName]
+	if !ok {
+		return "", fmt.Errorf("cluster name key %v not found", keyName)
+	}
+	return name, nil
+}
+
+func getConfigMap(ctx context.Context, configMap string) (map[string]string, error) {
+	rootPEM, err := os.ReadFile(filepath.Join(k8sServiceAccountPrefix, k8sCert))
+	if err != nil {
+		return nil, err
+	}
+	token, err := os.ReadFile(filepath.Join(k8sServiceAccountPrefix, k8sToken))
+	if err != nil {
+		return nil, err
+	}
+
+	roots := x509.NewCertPool()
+	ok := roots.AppendCertsFromPEM([]byte(rootPEM))
+	if !ok {
+		panic("failed to parse root certificate")
+	}
+	tr := &http.Transport{
+		TLSClientConfig: &tls.Config{
+			RootCAs: roots,
+		},
+	}
+	client := &http.Client{Transport: tr}
+	u := &url.URL{
+		Scheme: "https",
+		Host:   k8sAPIHost,
+		Path:   configMap,
+	}
+	req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Authorization", "Bearer "+string(token))
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	buf := bytes.NewBuffer(make([]byte, 0, 1024))
+	io.Copy(buf, resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("%s: ERROR: %v", buf.String(), resp.StatusCode)
+	}
+	cm := struct {
+		API  string            `json:"apiVersion"`
+		Data map[string]string `json:"data"`
+	}{}
+	if err := json.Unmarshal(buf.Bytes(), &cm); err != nil {
+		return nil, fmt.Errorf("%s: %v", buf.String(), err)
+	}
+	if cm.API != "v1" && len(cm.Data) == 0 {
+		return nil, fmt.Errorf("API version has changed to %v: found no config map data", cm.API)
+
+	}
+	return cm.Data, nil
+}
+
+func GetContainerID(cgroupFile string) (string, error) {
+	rd, err := os.Open(cgroupFile)
+	if err != nil {
+		return "", err
+	}
+	sc := bufio.NewScanner(rd)
+	cid := ""
+	for sc.Scan() {
+		line := sc.Text()
+		if l := len(line); l > kContainerIDLen {
+			cid = line[l-kContainerIDLen:]
+			break
+		}
+	}
+	if len(cid) == 0 {
+		return "", fmt.Errorf("failed to find a container id in %v", cgroupFile)
+	}
+	return cid, nil
+}