flink_exporter/main.go at develop · annaewolf/flink_exporter · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package main

import (
	"fmt"
	log "github.com/Sirupsen/logrus"
	"github.com/matsumana/flink_exporter/exporter"
	"github.com/prometheus/client_golang/prometheus"
	"gopkg.in/urfave/cli.v2"
	"net/http"
	"os"
)

const (
	version   = "2.3.0"
	endpoint  = "/metrics"
	namespace = "flink"
)

type appOpts struct {
	Name    string
	Version string
	Flags   []cli.Flag
}

func main() {
	opts := &appOpts{
		Name:    "flink_exporter",
		Version: version,
	}
	opts.Flags = []cli.Flag{
		&cli.StringFlag{
			Name:  "log-level",
			Usage: "Set Logging level",
			Value: "info",
		},
		&cli.IntFlag{
			Name:  "port",
			Usage: "The port number used to expose metrics via http",
			Value: 9160,
		},
		&cli.StringFlag{
			Name:  "flink-job-manager-url",
			Usage: "flink job manager url",
		},
		&cli.StringFlag{
			Name:  "yarn-resource-manager-url",
			Usage: "YARN ResourceManager url",
		},
	}
	log.Debugf("opts = %v", opts)

	err := newApp(opts).Run(os.Args)
	if err != nil {
		os.Exit(1)
	}
}

func newApp(opts *appOpts) *cli.App {
	return &cli.App{
		Name:    opts.Name,
		Version: opts.Version,
		Usage:   "Prometheus exporter for Apache Flink",
		Flags:   opts.Flags,
		Action:  action,
	}
}

func action(c *cli.Context) error {

	setupLogging(c)
	checkArgs(c)

	flinkJobManagerUrl := c.String("flink-job-manager-url")
	yarnResourceManagerUrl := c.String("yarn-resource-manager-url")

	// register exporter
	exporter := exporter.NewExporter(flinkJobManagerUrl, yarnResourceManagerUrl, namespace)
	prometheus.MustRegister(exporter)

	// http listen and serve
	port := c.Int("port")
	log.Debugf("port = %v", port)

	http.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
		w.Header().Add("Location", endpoint)
		w.WriteHeader(http.StatusMovedPermanently)
	})
	http.Handle(endpoint, prometheus.Handler())
	if err := http.ListenAndServe(fmt.Sprintf(":%d", port), nil); err != nil {
		log.Fatal(err)
	}

	return nil
}

func checkArgs(c *cli.Context) {
	flinkJobManagerUrl := c.String("flink-job-manager-url")
	yarnResourceManagerUrl := c.String("yarn-resource-manager-url")

	log.Debugf("flink-job-manager-url = %v", flinkJobManagerUrl)
	log.Debugf("yarn-resource-manager-url = %v", yarnResourceManagerUrl)

	if flinkJobManagerUrl == "" && yarnResourceManagerUrl == "" {
		log.Fatal("Specify either fink-job-manager-url or yarn-resource-manager-url. Can't specify both.")
	}

	if flinkJobManagerUrl != "" && yarnResourceManagerUrl != "" {
		log.Fatal("Specify either fink-job-manager-url or yarn-resource-manager-url. Can't specify both.")
	}
}

func setupLogging(c *cli.Context) {
	log.SetFormatter(&log.TextFormatter{
		FullTimestamp: true,
	})
	levelString := c.String("log-level")
	level, err := log.ParseLevel(levelString)
	if err != nil {
		log.Fatalf("could not set log level to '%s';err:<%s>", levelString, err)
	}
	log.SetLevel(level)
}