Gearman monitoring

cmd/gearmand/main.go

@@ -2,41 +2,30 @@
 package main
 
 import (
+	_ "net/http/pprof"
 	"os"
 
 	gearmand "github.com/appscode/g2/pkg/server"
-	"github.com/appscode/g2/pkg/storage"
-	"github.com/appscode/g2/pkg/storage/leveldb"
 	"github.com/appscode/go/flags"
 	"github.com/appscode/go/runtime"
-	"github.com/appscode/log"
 	logs "github.com/appscode/log/golog"
 	"github.com/spf13/pflag"
 )
 
-var (
-	addr       string
-	storageDir string
-)
-
 func main() {
-	pflag.StringVar(&addr, "addr", ":4730", "listening on, such as 0.0.0.0:4730")
-	pflag.StringVar(&storageDir, "storage-dir", os.TempDir(), "Directory where LevelDB file is stored.")
+	cfg := &gearmand.Config{}
+
+	pflag.StringVar(&cfg.ListenAddr, "addr", ":4730", "listening on, such as 0.0.0.0:4730")
+	pflag.StringVar(&cfg.Storage, "storage-dir", os.TempDir()+"/gearmand", "Directory where LevelDB file is stored.")
+	pflag.StringVar(&cfg.WebAddress, "web.addr", ":3000", "Server HTTP api Address")
+
+	defer runtime.HandleCrash()
 
 	flags.InitFlags()
 	logs.InitLogs()
 	defer logs.FlushLogs()
-	flags.DumpAll()
 
-	var store storage.Db
-	if storageDir != "" {
-		if s, err := leveldbq.New(storageDir); err == nil {
-			store = s
-		} else {
-			log.Info(err)
-		}
-	}
+	flags.DumpAll()
 
-	defer runtime.HandleCrash()
-	gearmand.NewServer(store).Start(addr)
+	gearmand.NewServer(cfg).Start()
 }

glide.yaml

@@ -1,12 +1,17 @@
 package: github.com/appscode/g2
 import:
-- package: gopkg.in/macaron.v1
-  version: v1.1.11
 - package: github.com/mikespook/golib
-- package: github.com/ngaut/stats
 - package: github.com/syndtr/goleveldb
 - package: github.com/appscode/go
 - package: github.com/appscode/log
 - package: github.com/spf13/pflag
 - package: gopkg.in/robfig/cron.v2
   version: v2
+- package: github.com/mikespook/golib
+- package: github.com/prometheus/client_golang
+  version: ^0.8.0
+- package: github.com/spf13/pflag
+testImport:
+- package: github.com/stretchr/testify
+  subpackages:
+  - assert

pkg/metrics/collector.go

@@ -0,0 +1,24 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+type collector struct {
+	metrics []*element
+}
+
+func (s *collector) Describe(ch chan<- *prometheus.Desc) {
+	for _, c := range s.metrics {
+		ch <- c.desc
+	}
+}
+
+func (s *collector) Collect(ch chan<- prometheus.Metric) {
+	for _, c := range s.metrics {
+		c.collect(c.desc, ch)
+	}
+}
+
+type element struct {
+	collect func(desc *prometheus.Desc, ch chan<- prometheus.Metric)
+	desc    *prometheus.Desc
+}

pkg/metrics/server_metrics.go

@@ -0,0 +1,119 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	serverNamespace = "gearman_server"
+)
+
+type ServerData interface {
+	Stats() map[string]int
+	Clients() int
+	Workers() int
+	Jobs() int
+	RunningJobsByWorker() map[string]int
+	RunningJobsByFunction() map[string]int
+}
+
+// TODO: Add Some More Complex Matrics As Needed
+func NewServerCollector(s ServerData) prometheus.Collector {
+	return &collector{
+		metrics: []*element{
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(serverNamespace, "", "worker_count"),
+					"Count Connected Workers",
+					nil, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					ch <- prometheus.MustNewConstMetric(
+						d,
+						prometheus.GaugeValue,
+						float64(s.Workers()),
+					)
+				},
+			},
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(serverNamespace, "", "job_count"),
+					"Count jobs",
+					nil, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					ch <- prometheus.MustNewConstMetric(
+						d,
+						prometheus.GaugeValue,
+						float64(s.Jobs()),
+					)
+				},
+			},
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(serverNamespace, "", "client_count"),
+					"Count Connected Clients",
+					nil, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					ch <- prometheus.MustNewConstMetric(
+						d,
+						prometheus.GaugeValue,
+						float64(s.Clients()),
+					)
+				},
+			},
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(serverNamespace, "", "worker_running_job_count"),
+					"Running Job By workers",
+					[]string{"worker"}, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					for k, v := range s.RunningJobsByWorker() {
+						ch <- prometheus.MustNewConstMetric(
+							d,
+							prometheus.GaugeValue,
+							float64(v),
+							k,
+						)
+					}
+				},
+			},
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(serverNamespace, "", "function_running_job_count"),
+					"Running job count by functions",
+					[]string{"function"}, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					for k, v := range s.RunningJobsByFunction() {
+						ch <- prometheus.MustNewConstMetric(
+							d,
+							prometheus.GaugeValue,
+							float64(v),
+							k,
+						)
+					}
+				},
+			},
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(serverNamespace, "", "stats"),
+					"Running job count by functions",
+					[]string{"stats"}, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					for k, v := range s.Stats() {
+						ch <- prometheus.MustNewConstMetric(
+							d,
+							prometheus.GaugeValue,
+							float64(v),
+							k,
+						)
+					}
+				},
+			},
+		},
+	}
+}

pkg/metrics/worker_metrics.go

@@ -0,0 +1,51 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	workerNamespace = "gearman_worker"
+)
+
+type WorkerData interface {
+	Running() (string, int)
+	Agents() int
+}
+
+func NewWorkerCollector(w WorkerData) prometheus.Collector {
+	return &collector{
+		metrics: []*element{
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(workerNamespace, "", "worker_running"),
+					"Worker Running",
+					[]string{"worker"}, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					id, running := w.Running()
+					ch <- prometheus.MustNewConstMetric(
+						d,
+						prometheus.GaugeValue,
+						float64(running),
+						id,
+					)
+				},
+			},
+			{
+				desc: prometheus.NewDesc(
+					prometheus.BuildFQName(workerNamespace, "", "agent_count"),
+					"Count Connected Agents",
+					nil, nil,
+				),
+				collect: func(d *prometheus.Desc, ch chan<- prometheus.Metric) {
+					ch <- prometheus.MustNewConstMetric(
+						d,
+						prometheus.GaugeValue,
+						float64(w.Agents()),
+					)
+				},
+			},
+		},
+	}
+}

pkg/runtime/protocol.go

@@ -1,6 +1,8 @@
 package runtime
 
-import "fmt"
+import (
+	"fmt"
+)
 
 /*
 Binary Packet