Merge pull request #98 from whywaita/fix/97

Set timeout in create an instance

go.mod

@@ -9,15 +9,14 @@ require (
 	github.com/google/go-github/v35 v35.2.0
 	github.com/hashicorp/go-plugin v1.4.0
 	github.com/jmoiron/sqlx v1.2.0
-	github.com/lestrrat-go/jwx v1.2.0 // indirect
+	github.com/lestrrat-go/jwx v1.2.0
 	github.com/ory/dockertest/v3 v3.6.2
-	github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
-	github.com/prometheus/client_golang v1.10.0 // indirect
+	github.com/patrickmn/go-cache v2.1.0+incompatible
+	github.com/prometheus/client_golang v1.10.0
 	github.com/satori/go.uuid v1.2.0
 	goji.io v2.0.2+incompatible
 	golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421
 	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
 	google.golang.org/grpc v1.33.2
-	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 // indirect
 	google.golang.org/protobuf v1.26.0
 )

pkg/metric/collector.go

@@ -0,0 +1,125 @@
+package metric
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/whywaita/myshoes/pkg/datastore"
+	"github.com/whywaita/myshoes/pkg/logger"
+)
+
+const (
+	namespace = "myshoes"
+)
+
+var (
+	scrapeDurationDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, "", "collector_duration_seconds"),
+		"Collector time duration.",
+		[]string{"collector"}, nil,
+	)
+)
+
+// Collector is a collector for prometheus
+type Collector struct {
+	ctx      context.Context
+	metrics  Metrics
+	ds       datastore.Datastore
+	scrapers []Scraper
+}
+
+// NewCollector create a collector
+func NewCollector(ctx context.Context, ds datastore.Datastore) *Collector {
+	return &Collector{
+		ctx:      ctx,
+		metrics:  NewMetrics(),
+		ds:       ds,
+		scrapers: NewScrapers(),
+	}
+}
+
+// Describe describe metrics
+func (c *Collector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.metrics.TotalScrapes.Desc()
+	ch <- c.metrics.Error.Desc()
+	c.metrics.ScrapeErrors.Describe(ch)
+}
+
+// Collect collect metrics
+func (c *Collector) Collect(ch chan<- prometheus.Metric) {
+	c.scrape(c.ctx, ch)
+
+	ch <- c.metrics.TotalScrapes
+	ch <- c.metrics.Error
+	c.metrics.ScrapeErrors.Collect(ch)
+}
+
+func (c *Collector) scrape(ctx context.Context, ch chan<- prometheus.Metric) {
+	c.metrics.TotalScrapes.Inc()
+	c.metrics.Error.Set(0)
+
+	var wg sync.WaitGroup
+	for _, scraper := range c.scrapers {
+		wg.Add(1)
+		go func(scraper Scraper) {
+			defer wg.Done()
+			label := fmt.Sprintf("collect.%s", scraper.Name())
+			scrapeStartTime := time.Now()
+			if err := scraper.Scrape(ctx, c.ds, ch); err != nil {
+				logger.Logf(false, "failed to scrape metrics (name: %s): %+v", scraper.Name(), err)
+				c.metrics.ScrapeErrors.WithLabelValues(label).Inc()
+				c.metrics.Error.Set(1)
+			}
+			ch <- prometheus.MustNewConstMetric(scrapeDurationDesc, prometheus.GaugeValue, time.Since(scrapeStartTime).Seconds(), label)
+		}(scraper)
+	}
+	wg.Wait()
+}
+
+// Scraper is interface for scraping
+type Scraper interface {
+	Name() string
+	Help() string
+	Scrape(ctx context.Context, ds datastore.Datastore, ch chan<- prometheus.Metric) error
+}
+
+// NewScrapers return list of scraper
+func NewScrapers() []Scraper {
+	return []Scraper{
+		ScraperDatastore{},
+	}
+}
+
+// Metrics is data in scraper
+type Metrics struct {
+	TotalScrapes prometheus.Counter
+	ScrapeErrors *prometheus.CounterVec
+	Error        prometheus.Gauge
+}
+
+// NewMetrics create a metrics
+func NewMetrics() Metrics {
+	return Metrics{
+		TotalScrapes: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "",
+			Name:      "scrapes_total",
+			Help:      "Total number of times myshoes was scraped for metrics.",
+		}),
+		ScrapeErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: "",
+			Name:      "scrape_errors_total",
+			Help:      "Total number of times an error occurred scraping a myshoes.",
+		}, []string{"collector"}),
+		Error: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: "",
+			Name:      "last_scrape_error",
+			Help:      "Whether the last scrape of metrics from myshoes resulted in an error (1 for error, 0 for success).",
+		}),
+	}
+}

pkg/metric/scrape_datastore.go

@@ -0,0 +1,96 @@
+package metric
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/whywaita/myshoes/pkg/datastore"
+)
+
+const datastoreName = "datastore"
+
+var (
+	datastoreJobsDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, datastoreName, "jobs"),
+		"Number of jobs",
+		[]string{"target_id"}, nil,
+	)
+	datastoreTargetsDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, datastoreName, "targets"),
+		"Number of targets",
+		[]string{"resource_type"}, nil,
+	)
+)
+
+// ScraperDatastore is scraper implement for datastore.Datastore
+type ScraperDatastore struct{}
+
+// Name return name
+func (ScraperDatastore) Name() string {
+	return datastoreName
+}
+
+// Help return help
+func (ScraperDatastore) Help() string {
+	return "Collect from datastore"
+}
+
+// Scrape scrape metrics
+func (ScraperDatastore) Scrape(ctx context.Context, ds datastore.Datastore, ch chan<- prometheus.Metric) error {
+	if err := scrapeJobs(ctx, ds, ch); err != nil {
+		return fmt.Errorf("failed to scrape jobs: %w", err)
+	}
+	if err := scrapeTargets(ctx, ds, ch); err != nil {
+		return fmt.Errorf("failed to scrape targets: %w", err)
+	}
+
+	return nil
+}
+
+func scrapeJobs(ctx context.Context, ds datastore.Datastore, ch chan<- prometheus.Metric) error {
+	jobs, err := ds.ListJobs(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to list jobs: %w", err)
+	}
+
+	if len(jobs) == 0 {
+		ch <- prometheus.MustNewConstMetric(
+			datastoreJobsDesc, prometheus.GaugeValue, 0, "none",
+		)
+		return nil
+	}
+
+	result := map[string]float64{} // key: target_id, value: number
+	for _, j := range jobs {
+		result[j.TargetID.String()]++
+	}
+	for targetID, number := range result {
+		ch <- prometheus.MustNewConstMetric(
+			datastoreJobsDesc, prometheus.GaugeValue, number, targetID,
+		)
+	}
+
+	return nil
+}
+
+func scrapeTargets(ctx context.Context, ds datastore.Datastore, ch chan<- prometheus.Metric) error {
+	targets, err := datastore.ListTargets(ctx, ds)
+	if err != nil {
+		return fmt.Errorf("failed to list targets: %w", err)
+	}
+
+	result := map[string]float64{} // key: resource_type, value: number
+	for _, t := range targets {
+		result[t.ResourceType.String()]++
+	}
+	for rt, number := range result {
+		ch <- prometheus.MustNewConstMetric(
+			datastoreTargetsDesc, prometheus.GaugeValue, number, rt,
+		)
+	}
+
+	return nil
+}
+
+var _ Scraper = ScraperDatastore{}

pkg/starter/starter.go

@@ -97,7 +97,9 @@ func (s *Starter) do(ctx context.Context) error {
 				return
 			}
 
-			cloudID, ipAddress, shoesType, err := s.bung(ctx, job.UUID, *target)
+			cctx, cancel := context.WithTimeout(ctx, runner.MustRunningTime)
+			defer cancel()
+			cloudID, ipAddress, shoesType, err := s.bung(cctx, job.UUID, *target)
 			if err != nil {
 				logger.Logf(false, "failed to bung (target ID: %s, job ID: %s): %+v\n", job.TargetID, job.UUID, err)
 
@@ -160,7 +162,6 @@ func (s *Starter) do(ctx context.Context) error {
 			}
 		}()
 	}
-
 	wg.Wait()
 
 	return nil

pkg/web/http.go

@@ -7,8 +7,6 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-
 	"github.com/whywaita/myshoes/internal/config"
 	"github.com/whywaita/myshoes/pkg/datastore"
 	"github.com/whywaita/myshoes/pkg/logger"
@@ -34,7 +32,6 @@ func NewMux(ds datastore.Datastore) *goji.Mux {
 		json.NewEncoder(w).Encode(h)
 		return
 	})
-	mux.Handle(pat.Get("/metrics"), promhttp.Handler())
 
 	mux.HandleFunc(pat.Post("/github/events"), func(w http.ResponseWriter, r *http.Request) {
 		apacheLogging(r)
@@ -73,6 +70,12 @@ func NewMux(ds datastore.Datastore) *goji.Mux {
 		handleConfigStrict(w, r)
 	})
 
+	// metrics endpoint
+	mux.HandleFunc(pat.Get("/metrics"), func(w http.ResponseWriter, r *http.Request) {
+		apacheLogging(r)
+		handleMetrics(w, r, ds)
+	})
+
 	return mux
 }
 

pkg/web/metrics.go

@@ -0,0 +1,27 @@
+package web
+
+import (
+	"net/http"
+
+	"github.com/whywaita/myshoes/pkg/datastore"
+
+	"github.com/whywaita/myshoes/pkg/metric"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+func handleMetrics(w http.ResponseWriter, r *http.Request, ds datastore.Datastore) {
+	ctx := r.Context()
+
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(metric.NewCollector(ctx, ds))
+
+	gatherers := prometheus.Gatherers{
+		prometheus.DefaultGatherer,
+		registry,
+	}
+	h := promhttp.HandlerFor(gatherers, promhttp.HandlerOpts{})
+	h.ServeHTTP(w, r)
+	return
+}