软件测试 | 普罗米修斯 - 自定义 exporter

import timefrom prometheus_client.core import GaugeMetricFamily, REGISTRY, CounterMetricFamilyfrom prometheus_client import start_http_server

class CustomCollector(object):    def __init__(self):        pass
    def collect(self):        g = GaugeMetricFamily("MemoryUsage", 'Help text', labels=['instance'])        g.add_metric(["instance01.us.west.local"], 20)        yield g
        c = CounterMetricFamily("HttpRequests", 'Help text', labels=['app'])        c.add_metric(["example"], 2000)        yield c

if __name__ == '__main__':    start_http_server(8000)    REGISTRY.register(CustomCollector())

<!-- The client --><dependency>    <groupId>io.prometheus</groupId>    <artifactId>simpleclient</artifactId>    <version>0.6.0</version></dependency>
<dependency>    <groupId>io.prometheus</groupId>    <artifactId>simpleclient_httpserver</artifactId>    <version>0.6.0</version></dependency>

package exporter;
import io.prometheus.client.Counter;import io.prometheus.client.exporter.HTTPServer;
import java.io.IOException;import java.util.Random;import java.util.concurrent.Executors;import java.util.concurrent.ScheduledExecutorService;import java.util.concurrent.TimeUnit;
public class CustomExporter {        # 注册一个counter类型的监控指标并带有一个名字叫method的label    static final Counter requests = Counter.build()            .name("my_library_requests_total").help("Total requests.")            .labelNames("method").register();
    public static void processGetRequest() {        requests.labels("get").inc();        // 在这编写监控逻辑    }
    public static void main(String[] args) throws IOException {        // 启动一个线程池，每隔10s种触发一次调用监控逻辑        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);        service.scheduleWithFixedDelay(() -> {            // 模拟一个监控动作， 调用counter类型的自增方法。 当然也可以调用上面的processGetRequest方法            double value = requests.labels("get").get();            System.out.println(value);            requests.labels("get").inc();        }, 0, 10, TimeUnit.SECONDS);
        // 利用普罗米修斯提供的httpserver 启动服务        HTTPServer server = new HTTPServer(1234);

    }}

module prophet-container-exporter
go 1.13
require (   github.com/pkg/errors v0.9.1   github.com/prometheus/client_golang v1.5.1   github.com/sirupsen/logrus v1.4.2   k8s.io/api v0.0.0-20190620084959-7cf5895f2711   k8s.io/apimachinery v0.0.0-20190612205821-1799e75a0719   k8s.io/client-go v0.0.0-20190620085101-78d2af792bab)

func init() {   log.SetOutput(os.Stdout)   log.Info("init the kubeconfig")   // 初始化k8s的client有两种方式。 如果当前是在pod中运行的话， client自己会找到容器对应目录下的service account信息与k8s的apiserver通信鉴权。 如果程序在集群外部运行， 那么需要给client提供一个kubeconfig文件   if isExist, err := PathExists("kubeconfig"); err != nil {      panic(err)   } else {      if isExist {         log.Info("now out of k8s cluster")         kubeConfig, err = clientcmd.BuildConfigFromFlags("", "kubeconfig")      } else {         log.Info("now In k8s cluster")         kubeConfig, err = rest.InClusterConfig()      }      if err != nil {         log.Error("cannot init the kubeconfig")         panic(err.Error())      }   }   var err error   k8s, err = kubernetes.NewForConfig(kubeConfig)   if err != nil {      log.Error("cannot init the k8s client")      panic(err.Error())   }   log.Info("init the k8sclient done, now begin to monitor the k8s")

   // 在开始监控之前首先遍历当前k8s集群中所有的pod，为每个容器建立一个guage类型的监控指标并记录在一个map中，方便后面程序根据这些容器执行具体的监控逻辑。   register := func() {      namespaceList, err := k8s.CoreV1().Namespaces().List(metav1.ListOptions{})      if err != nil {         log.Error(err)         os.Exit(1)      }
      for _, n := range namespaceList.Items {         namespace := n.Name         podList, err := k8s.CoreV1().Pods(namespace).List(metav1.ListOptions{})         if err != nil {            panic(errors.Wrapf(err, "cannot list k8s with namespace %s", namespace))         }         // 遍历所有pod         for _, pod := range podList.Items {                        # 只监控Running状态的Pod            if pod.Status.Phase != "Running" {               continue            }
            // 遍历pod下的容器, 每个容器的指标有3个label。1. 名称空间 2. pod名称  3. 容器名称            for _, container := range pod.Status.ContainerStatuses {               sdGauge := prometheus.NewGauge(prometheus.GaugeOpts{                  Name: "namespace_container_Socket_Close_Wait",                  Help: "num of socket with CLOSE-WAIT status in container",                  ConstLabels: map[string]string{                     "namespace": namespace,                     "pod":       pod.Name,                     "container": container.Name,                  },               })               if _, ok := sdMetrics[pod.Name+","+namespace]; !ok {                  prometheus.MustRegister(sdGauge)                  sdMetrics[pod.Name+","+namespace] = sdGauge               }            }         }      }
   }
   // 上来就注册一次   register()
   // 周期性注册namespace下所有pod中app容器的指标, 因为随时会有新的pod启动   go func() {      for {         time.Sleep(time.Minute * 10)         register()      }   }()}

func main() {
   // 周期性的获取最新的监控数据。 遍历所有容器， 向容器发送`cat /proc/net/tcp | awk '{print $4}'|grep 08|wc -l` 命令，该命令可以查询容器中socket的状态，计算一共有多少个处于CLOSE-WAIT的socket。而我们的监控目的就是通过这个指标判断是否存在socket泄露.   go func() {      for {         for podInfo, guage := range sdMetrics {            tmp := strings.Split(podInfo, ",")            podName := tmp[0]            namespace := tmp[1]
            log.WithFields(log.Fields{               "namespace": namespace,               "pod":       podName,            })            pod, _ := k8s.CoreV1().Pods(namespace).Get(podName, metav1.GetOptions{})
            for _, container := range pod.Status.ContainerStatuses {               commands := []string{"sh", "-c", "cat /proc/net/tcp | awk '{print $4}'|grep 08|wc -l"}               output, err := promethues.Exec(k8s, kubeConfig, commands, namespace, podName, container.Name)               if err != nil {                  log.Error(err.Error())                  continue               }               closeWait, err := strconv.ParseFloat(strings.Replace(output, "\n", "", -1), 32)
               if err != nil {                  fmt.Fprintf(os.Stdout, "err %s\n", errors.Wrap(err, "cannot trans string to float"))                  continue               }               guage.Set(closeWait)               log.Infof("successfully collect %s's socket status", podName)            }         }         time.Sleep(time.Minute * 10)      }   }()
   http.Handle("/metrics", promhttp.Handler())   log.Fatal(http.ListenAndServe("0.0.0.0:80", nil))
}