package inject import ( "fmt" "math" "strconv" "strings" "github.com/aws/aws-sdk-go/aws" "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" ) const envoyTracingConfigVolumeName = "envoy-tracing-config" // Envoy template variables used by envoys in pod and the envoy in VirtualGateway // as we use the same envoy image type EnvoyTemplateVariables struct { AWSRegion string MeshName string VirtualGatewayOrNodeName string Preview string EnableSDS bool SdsUdsPath string LogLevel string AdminAccessPort int32 AdminAccessLogFile string PreStopDelay string PostStartTimeout int32 PostStartInterval int32 SidecarImageRepository string SidecarImageTag string EnableXrayTracing bool XrayDaemonPort int32 XraySamplingRate string EnableJaegerTracing bool JaegerPort string JaegerAddress string EnableDatadogTracing bool DatadogTracerPort int32 DatadogTracerAddress string EnableStatsTags bool EnableStatsD bool StatsDPort int32 StatsDAddress string StatsDSocketPath string K8sVersion string ControllerVersion string EnableAdminAccessForIpv6 bool UseDualStackEndpoint string WaitUntilProxyReady bool UseFipsEndpoint string AwsAccessKeyId string AwsSecretAccessKey string AwsSessionToken string } func updateEnvMapForEnvoy(vars EnvoyTemplateVariables, env map[string]string, vname string) error { // add all the controller managed env to the map so // 1) we remove duplicates // 2) we don't allow overriding controller managed env with pod annotations env["APPMESH_VIRTUAL_NODE_NAME"] = vname env["AWS_REGION"] = vars.AWSRegion // For usage outside traditional EC2 / Fargate IAM based profiles, this is needed to // propagate permissions to envoy. This is a rare use-case that's mostly just for testing. if len(vars.AwsAccessKeyId) > 0 { env["AWS_ACCESS_KEY_ID"] = vars.AwsAccessKeyId } if len(vars.AwsSecretAccessKey) > 0 { env["AWS_SECRET_ACCESS_KEY"] = vars.AwsSecretAccessKey } if len(vars.AwsSessionToken) > 0 { env["AWS_SESSION_TOKEN"] = vars.AwsSessionToken } env["ENVOY_ADMIN_ACCESS_ENABLE_IPV6"] = strconv.FormatBool(vars.EnableAdminAccessForIpv6) env["APPMESH_DUALSTACK_ENDPOINT"] = vars.UseDualStackEndpoint env["APPMESH_FIPS_ENDPOINT"] = vars.UseFipsEndpoint // Set the value to 1 to connect to the App Mesh Preview Channel endpoint. // See https://docs.aws.amazon.com/app-mesh/latest/userguide/preview.html env["APPMESH_PREVIEW"] = vars.Preview // Specifies the log level for the Envoy container // Valid values: trace, debug, info, warning, error, critical, off env["ENVOY_LOG_LEVEL"] = vars.LogLevel if vars.EnableSDS { env["APPMESH_SDS_SOCKET_PATH"] = vars.SdsUdsPath } if vars.AdminAccessPort != 0 { // Specify a custom admin port for Envoy to listen on // Default: 9901 env["ENVOY_ADMIN_ACCESS_PORT"] = strconv.Itoa(int(vars.AdminAccessPort)) } if vars.AdminAccessLogFile != "" { // Specify a custom path to write Envoy access logs to // Default: /tmp/envoy_admin_access.log env["ENVOY_ADMIN_ACCESS_LOG_FILE"] = vars.AdminAccessLogFile } if vars.EnableXrayTracing { // Enables X-Ray tracing using 127.0.0.1:2000 as the default daemon endpoint // To enable, set the value to 1 env["ENABLE_ENVOY_XRAY_TRACING"] = "1" // Specify a port value to override the default X-Ray daemon port: 2000 env["XRAY_DAEMON_PORT"] = strconv.Itoa(int(vars.XrayDaemonPort)) // Override the default sampling rate of 0.05 (5%) for AWS X-Ray tracer // The value should be specified as a decimal between 0 and 1.00 (100%) samplingRate, ok := env["XRAY_SAMPLING_RATE"] if ok { // `podAnnotations` contains the sampling rate and gets preference over helm configuration // For now delete this value from env so that we can validate before adding again delete(env, "XRAY_SAMPLING_RATE") } else { // `podAnnotations` doesn't contain the sampling rate so get value from helm configuration samplingRate = vars.XraySamplingRate } fixedRate, err := strconv.ParseFloat(samplingRate, 32) if err != nil || float64(0) > fixedRate || float64(1) < fixedRate { // The value is not a decimal between 0 and 1.00 return errors.Errorf("tracing.samplingRate should be a decimal between 0 & 1.00, "+ "but instead got %s %v", samplingRate, err) } else { fixedRate = math.Round(fixedRate*100) / 100 env["XRAY_SAMPLING_RATE"] = strconv.FormatFloat(fixedRate, 'f', -1, 32) } } if vars.EnableDatadogTracing { // Enables Datadog trace collection using 127.0.0.1:8126 // as the default Datadog agent endpoint. To enable, set the value to 1 env["ENABLE_ENVOY_DATADOG_TRACING"] = "1" // Specify a port value to override the default Datadog agent port: 8126 env["DATADOG_TRACER_PORT"] = strconv.Itoa(int(vars.DatadogTracerPort)) // Specify an IP address or hostname to override the default Datadog agent address: 127.0.0.1 env["DATADOG_TRACER_ADDRESS"] = vars.DatadogTracerAddress } if vars.EnableStatsTags { env["ENABLE_ENVOY_STATS_TAGS"] = "1" } if vars.EnableStatsD { // Enables DogStatsD stats using 127.0.0.1:8125 // as the default daemon endpoint. To enable, set the value to 1 env["ENABLE_ENVOY_DOG_STATSD"] = "1" // Specify a port value to override the default DogStatsD daemon port. // This value will be overridden if `STATSD_SOCKET_PATH` is specified. env["STATSD_PORT"] = strconv.Itoa(int(vars.StatsDPort)) // Specify an IP address value to override the default DogStatsD daemon IP address // Default: 127.0.0.1. This variable can only be used with version 1.15.0 or later // of the Envoy image. This value will be overridden if `STATSD_SOCKET_PATH` is specified. env["STATSD_ADDRESS"] = vars.StatsDAddress // Specify a unix domain socket for DogStatsD daemon. If not specified and if DogStatsD // is enabled then defaults to DogStatsD daemon IP address port [default: 127.0.0.1:8125]. // This variable can only be used with version v1.19.1 or later. if statsDSocketPath := strings.TrimSpace(vars.StatsDSocketPath); statsDSocketPath != "" { env["STATSD_SOCKET_PATH"] = statsDSocketPath } } if vars.EnableJaegerTracing { env["ENABLE_ENVOY_JAEGER_TRACING"] = "1" env["JAEGER_TRACER_PORT"] = vars.JaegerPort env["JAEGER_TRACER_ADDRESS"] = vars.JaegerAddress } env["APPMESH_PLATFORM_K8S_VERSION"] = vars.K8sVersion env["APPMESH_PLATFORM_APP_MESH_CONTROLLER_VERSION"] = vars.ControllerVersion env["APPNET_AGENT_ADMIN_MODE"] = "uds" env["APPNET_AGENT_ADMIN_UDS_PATH"] = "/tmp/agent.sock" return nil } func buildEnvoySidecar(vars EnvoyTemplateVariables, env map[string]string) (corev1.Container, error) { envoy := corev1.Container{ Name: "envoy", Image: fmt.Sprintf("%s:%s", vars.SidecarImageRepository, vars.SidecarImageTag), SecurityContext: &corev1.SecurityContext{ RunAsUser: aws.Int64(1337), }, Ports: []corev1.ContainerPort{ { Name: "stats", ContainerPort: vars.AdminAccessPort, Protocol: "TCP", }, }, Lifecycle: &corev1.Lifecycle{ PostStart: nil, PreStop: &corev1.LifecycleHandler{ Exec: &corev1.ExecAction{Command: []string{ "sh", "-c", fmt.Sprintf("sleep %s", vars.PreStopDelay), }}, }, }, } if vars.WaitUntilProxyReady { envoy.Lifecycle.PostStart = &corev1.LifecycleHandler{ Exec: &corev1.ExecAction{Command: []string{ // use bash regex and rematch to parse and check envoy version is >= 1.22.2.1 "sh", "-c", fmt.Sprintf("if [[ $(/usr/bin/envoy --version) =~ ([0-9]+)\\.([0-9]+)\\.([0-9]+)-appmesh\\.([0-9]+) && "+ "${BASH_REMATCH[1]} -ge 2 || (${BASH_REMATCH[1]} -ge 1 && ${BASH_REMATCH[2]} -gt 22) || (${BASH_REMATCH[1]} -ge 1 && "+ "${BASH_REMATCH[2]} -ge 22 && ${BASH_REMATCH[3]} -gt 2) || (${BASH_REMATCH[1]} -ge 1 && ${BASH_REMATCH[2]} -ge 22 && "+ "${BASH_REMATCH[3]} -ge 2 && ${BASH_REMATCH[4]} -gt 0) ]]; then APPNET_AGENT_POLL_ENVOY_READINESS_TIMEOUT_S=%d "+ "APPNET_AGENT_POLL_ENVOY_READINESS_INTERVAL_S=%d /usr/bin/agent -envoyReadiness; else echo 'WaitUntilProxyReady "+ "is not supported in Envoy version < 1.22.2.1'; fi", vars.PostStartTimeout, vars.PostStartInterval), }}, } } vname := fmt.Sprintf("mesh/%s/virtualNode/%s", vars.MeshName, vars.VirtualGatewayOrNodeName) if err := updateEnvMapForEnvoy(vars, env, vname); err != nil { return envoy, err } envoy.Env = getEnvoyEnv(env) return envoy, nil } func getEnvoyEnv(env map[string]string) []corev1.EnvVar { ev := []corev1.EnvVar{} for key, val := range env { switch k := key; k { case "STATSD_ADDRESS", "DATADOG_TRACER_ADDRESS": if val == "ref:status.hostIP" { ev = append(ev, refHostIP(key)) } else { ev = append(ev, envVar(key, val)) } default: ev = append(ev, envVar(key, val)) } } ev = append(ev, refPodUid("APPMESH_PLATFORM_K8S_POD_UID")) return ev } func envoyReadinessProbe(initialDelaySeconds int32, periodSeconds int32, adminAccessPort string) *corev1.Probe { envoyReadinessCommand := "curl -s http://localhost:" + adminAccessPort + "/server_info | grep state | grep -q LIVE" return &corev1.Probe{ ProbeHandler: corev1.ProbeHandler{ // server_info returns the following struct: // { // "version": "...", // "state": "...", // "uptime_current_epoch": "{...}", // "uptime_all_epochs": "{...}", // "hot_restart_version": "...", // "command_line_options": "{...}" // } // server_info->state supports the following states: LIVE, DRAINING, PRE_INITIALIZING, and INITIALIZING // LIVE: Server is live and serving traffic // DRAINING: Server is draining listeners in response to external health checks failing // PRE_INITIALIZING: Server has not yet completed cluster manager initialization // INITIALIZING: Server is running the cluster manager initialization callbacks Exec: &corev1.ExecAction{Command: []string{ "sh", "-c", envoyReadinessCommand, }}, }, // Number of seconds after the container has started before readiness probes are initiated InitialDelaySeconds: initialDelaySeconds, // Number of seconds after which the probe times out // This is a call to the local Envoy endpoint. 1 second is more than enough for timeout TimeoutSeconds: 1, // How often (in seconds) to perform the probe PeriodSeconds: periodSeconds, // Minimum consecutive successes for the probe to be considered successful after having failed // If Envoy shows LIVE status once, we're good to call it a success SuccessThreshold: 1, // Minimum consecutive failures for the probe to be considered failed after having succeeded // Keeping the failure threshold to 3 to not fail preemptively FailureThreshold: 3, } } func sidecarResources(cpuRequest, memoryRequest, cpuLimit, memoryLimit string) (corev1.ResourceRequirements, error) { resources := corev1.ResourceRequirements{} if cpuRequest != "" || memoryRequest != "" { requests := corev1.ResourceList{} if cpuRequest != "" { cr, err := resource.ParseQuantity(cpuRequest) if err != nil { return resources, err } requests["cpu"] = cr } if memoryRequest != "" { mr, err := resource.ParseQuantity(memoryRequest) if err != nil { return resources, err } requests["memory"] = mr } resources.Requests = requests } if cpuLimit != "" || memoryLimit != "" { limits := corev1.ResourceList{} if cpuLimit != "" { cl, err := resource.ParseQuantity(cpuLimit) if err != nil { return resources, err } limits["cpu"] = cl } if memoryLimit != "" { ml, err := resource.ParseQuantity(memoryLimit) if err != nil { return resources, err } limits["memory"] = ml } resources.Limits = limits } return resources, nil } // refHostIP is to use the k8s downward API and render the host IP // this is useful in cases where the tracing agent is running as a daemonset func refHostIP(envName string) corev1.EnvVar { return corev1.EnvVar{ Name: envName, Value: "", ValueFrom: &corev1.EnvVarSource{ FieldRef: &corev1.ObjectFieldSelector{ FieldPath: "status.hostIP", }, }, } } func envVar(envName, envVal string) corev1.EnvVar { return corev1.EnvVar{ Name: envName, Value: envVal, } } // refPodUid is to use the k8s downward API and render pod uid // this info will be used to help App Mesh team to identify // the platform Envoy is running on func refPodUid(envName string) corev1.EnvVar { return corev1.EnvVar{ Name: envName, Value: "", ValueFrom: &corev1.EnvVarSource{ FieldRef: &corev1.ObjectFieldSelector{ FieldPath: "metadata.uid", }, }, } } // containsEnvoyTracingConfigVolume checks whether pod already contains "envoy-tracing-config" volume func containsEnvoyTracingConfigVolume(pod *corev1.Pod) bool { for _, volume := range pod.Spec.Volumes { if volume.Name == envoyTracingConfigVolumeName { return true } } return false }