linkerd2/controller/heartbeat/heartbeat.go

package heartbeat

import (
	"context"
	"fmt"
	"io/ioutil"
	"math"
	"net/http"
	"net/url"
	"strconv"
	"time"

	"github.com/linkerd/linkerd2/pkg/healthcheck"
	"github.com/linkerd/linkerd2/pkg/k8s"
	"github.com/linkerd/linkerd2/pkg/version"
	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"github.com/prometheus/common/model"
	log "github.com/sirupsen/logrus"
)

// K8sValues gathers relevant heartbeat information from Kubernetes
func K8sValues(kubeAPI *k8s.KubernetesAPI, controlPlaneNamespace string) url.Values {
	v := url.Values{}

	cm, configPB, err := healthcheck.FetchLinkerdConfigMap(kubeAPI, controlPlaneNamespace)
	if err != nil {
		log.Errorf("Failed to fetch linkerd-config: %s", err)
	} else {
		v.Set("uuid", configPB.GetInstall().GetUuid())
		v.Set("install-time", strconv.FormatInt(cm.GetCreationTimestamp().Unix(), 10))
	}

	versionInfo, err := kubeAPI.GetVersionInfo()
	if err != nil {
		log.Errorf("Failed to fetch Kubernetes version info: %s", err)
	} else {
		v.Set("k8s-version", versionInfo.String())
	}

	return v
}

// PromValues gathers relevant heartbeat information from Prometheus
func PromValues(promAPI promv1.API, controlPlaneNamespace string) url.Values {
	v := url.Values{}

	jobProxyLabels := model.LabelSet{"job": "linkerd-proxy"}

	// total-rps
	query := fmt.Sprintf("sum(rate(request_total%s[30s]))", jobProxyLabels.Merge(model.LabelSet{"direction": "inbound"}))
	value, err := promQuery(promAPI, query, 0)
	if err != nil {
		log.Errorf("Prometheus query failed: %s", err)
	} else {
		v.Set("total-rps", value)
	}

	// meshed-pods
	query = fmt.Sprintf("count(count by (pod) (request_total%s))", jobProxyLabels)
	value, err = promQuery(promAPI, query, 0)
	if err != nil {
		log.Errorf("Prometheus query failed: %s", err)
	} else {
		v.Set("meshed-pods", value)
	}

	// p95-handle-us
	query = fmt.Sprintf("histogram_quantile(0.99, sum(rate(request_handle_us_bucket%s[24h])) by (le))", jobProxyLabels)
	value, err = promQuery(promAPI, query, 0)
	if err != nil {
		log.Errorf("Prometheus query failed: %s", err)
	} else {
		v.Set("p99-handle-us", value)
	}

	// container metrics
	for _, container := range []struct {
		name model.LabelValue
		ns   model.LabelValue
	}{
		{
			name: "linkerd-proxy",
		},
		{
			name: "destination",
			ns:   "linkerd",
		},
		{
			name: "prometheus",
			ns:   "linkerd",
		},
	} {
		containerLabels := model.LabelSet{
			"job":            "kubernetes-nodes-cadvisor",
			"container_name": container.name,
		}
		if container.ns != "" {
			containerLabels["namespace"] = container.ns
		}

		// max-mem
		query = fmt.Sprintf("max(container_memory_working_set_bytes%s)", containerLabels)
		value, err = promQuery(promAPI, query, 0)
		if err != nil {
			log.Errorf("Prometheus query failed: %s", err)
		} else {
			param := fmt.Sprintf("max-mem-%s", container.name)
			v.Set(param, value)
		}

		// p95-cpu
		query = fmt.Sprintf("max(quantile_over_time(0.95,rate(container_cpu_usage_seconds_total%s[5m])[24h:5m]))", containerLabels)
		value, err = promQuery(promAPI, query, 3)
		if err != nil {
			log.Errorf("Prometheus query failed: %s", err)
		} else {
			param := fmt.Sprintf("p95-cpu-%s", container.name)
			v.Set(param, value)
		}
	}

	return v
}

func promQuery(promAPI promv1.API, query string, precision int) (string, error) {
	log.Debugf("Prometheus query: %s", query)

	res, err := promAPI.Query(context.Background(), query, time.Time{})
	if err != nil {
		return "", err
	}

	switch result := res.(type) {
	case model.Vector:
		if len(result) != 1 {
			return "", fmt.Errorf("unexpected result Prometheus result vector length: %d", len(result))
		}
		f := float64(result[0].Value)
		if math.IsNaN(f) {
			return "", fmt.Errorf("unexpected sample value: %v", result[0].Value)
		}

		return strconv.FormatFloat(f, 'f', precision, 64), nil
	}

	return "", fmt.Errorf("unexpected query result type (expected Vector): %s", res.Type())
}

// MergeValues merges two url.Values
func MergeValues(v1, v2 url.Values) url.Values {
	v := url.Values{}
	for key, val := range v1 {
		v[key] = val
	}
	for key, val := range v2 {
		v[key] = val
	}
	return v
}

// Send takes a map of url.Values and sends them to versioncheck.linkerd.io
func Send(v url.Values) error {
	return send(http.DefaultClient, version.CheckURL, v)
}

func send(client *http.Client, baseURL string, v url.Values) error {
	req, err := http.NewRequest("GET", baseURL, nil)
	if err != nil {
		return fmt.Errorf("failed to create HTTP request for base URL [%s]: %s", baseURL, err)
	}
	req.URL.RawQuery = v.Encode()

	log.Infof("Sending heartbeat: %s", req.URL.String())
	resp, err := client.Do(req)
	if err != nil {
		return fmt.Errorf("Check URL [%s] request failed with: %s", req.URL.String(), err)
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("failed to read response body: %s", err)
	}

	log.Infof("Successfully sent heartbeat: %s", string(body))

	return nil
}