Add k8s scraper service

server/server.go

@@ -233,10 +233,6 @@ func New(c *Config, buildInfo BuildInfo, logService logging.Interface) (*Server,
 	s.appendReplayService()
 
 	// Append third-party integrations
-	if err := s.appendK8sService(); err != nil {
-		return nil, errors.Wrap(err, "kubernetes service")
-	}
-
 	// Append extra input services
 	s.appendCollectdService()
 	s.appendUDPServices()
@@ -254,6 +250,11 @@ func New(c *Config, buildInfo BuildInfo, logService logging.Interface) (*Server,
 
 	// Append Scraper and discovery services
 	s.appendScraperService()
+
+	if err := s.appendK8sService(s.ScraperService); err != nil {
+		return nil, errors.Wrap(err, "kubernetes service")
+	}
+
 	s.appendAzureService(s.ScraperService)
 	s.appendConsulService(s.ScraperService)
 	s.appendDNSService(s.ScraperService)
@@ -415,10 +416,10 @@ func (s *Server) appendReplayService() {
 	s.AppendService("replay", srv)
 }
 
-func (s *Server) appendK8sService() error {
+func (s *Server) appendK8sService(r scraper.Registry) error {
 	c := s.config.Kubernetes
 	l := s.LogService.NewLogger("[kubernetes] ", log.LstdFlags)
-	srv, err := k8s.NewService(c, l)
+	srv, err := k8s.NewService(c, r, l)
 	if err != nil {
 		return err
 	}

server/server_test.go

@@ -6154,6 +6154,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 						"in-cluster":  false,
 						"namespace":   "",
 						"token":       false,
+						"resource":    "",
 					},
 					Redacted: []string{
 						"token",
@@ -6170,6 +6171,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 					"in-cluster":  false,
 					"namespace":   "",
 					"token":       false,
+					"resource":    "",
 				},
 				Redacted: []string{
 					"token",
@@ -6194,6 +6196,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 								"in-cluster":  false,
 								"namespace":   "",
 								"token":       true,
+								"resource":    "",
 							},
 							Redacted: []string{
 								"token",
@@ -6210,6 +6213,7 @@ func TestServer_UpdateConfig(t *testing.T) {
 							"in-cluster":  false,
 							"namespace":   "",
 							"token":       true,
+							"resource":    "",
 						},
 						Redacted: []string{
 							"token",

services/k8s/config.go

@@ -4,10 +4,12 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"io/ioutil"
+	"net/url"
 
 	"github.com/influxdata/kapacitor/listmap"
 	"github.com/influxdata/kapacitor/services/k8s/client"
 	"github.com/pkg/errors"
+	"github.com/prometheus/prometheus/config"
 )
 
 type Config struct {
@@ -18,6 +20,7 @@ type Config struct {
 	Token      string   `toml:"token" override:"token,redact"`
 	CAPath     string   `toml:"ca-path" override:"ca-path"`
 	Namespace  string   `toml:"namespace" override:"namespace"`
+	Resource   string   `toml:"resource" override:"resource"`
 }
 
 func NewConfig() Config {
@@ -43,6 +46,21 @@ func (c Config) Validate() error {
 		}
 	} else if len(c.APIServers) == 0 {
 		return errors.New("no api-servers specified, must provide at least one server URL")
+	} else {
+		for _, s := range c.APIServers {
+			_, err := url.Parse(s)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	if c.Resource != "" {
+		switch c.Resource {
+		case "node", "pod", "service", "endpoint":
+		default:
+			return errors.New("Resource must be one of node, pod, service, or endpoints")
+		}
+
 	}
 	return nil
 }
@@ -92,3 +110,45 @@ func (cs Configs) Validate() error {
 	}
 	return nil
 }
+
+// Prom writes the prometheus configuration for discoverer into ScrapeConfig
+func (c Config) Prom(conf *config.ScrapeConfig) {
+	if len(c.APIServers) == 0 {
+		conf.ServiceDiscoveryConfig.KubernetesSDConfigs = []*config.KubernetesSDConfig{
+			&config.KubernetesSDConfig{
+				Role:        config.KubernetesRole(c.Resource),
+				BearerToken: c.Token,
+				TLSConfig: config.TLSConfig{
+					CAFile: c.CAPath,
+				},
+			},
+		}
+		return
+	}
+
+	sds := make([]*config.KubernetesSDConfig, len(c.APIServers))
+	for i, srv := range c.APIServers {
+		url, _ := url.Parse(srv)
+		sds[i] = &config.KubernetesSDConfig{
+			APIServer: config.URL{
+				URL: url,
+			},
+			Role:        config.KubernetesRole(c.Resource),
+			BearerToken: c.Token,
+			TLSConfig: config.TLSConfig{
+				CAFile: c.CAPath,
+			},
+		}
+	}
+	conf.ServiceDiscoveryConfig.KubernetesSDConfigs = sds
+}
+
+// Service return discoverer type
+func (c Config) Service() string {
+	return "kubernetes"
+}
+
+// ServiceID returns the discoverers name
+func (c Config) ServiceID() string {
+	return c.ID
+}

services/k8s/service.go

@@ -6,15 +6,20 @@ import (
 	"sync"
 
 	"github.com/influxdata/kapacitor/services/k8s/client"
+	"github.com/influxdata/kapacitor/services/scraper"
 )
 
+// Service is the kubernetes discovery and autoscale service
 type Service struct {
 	mu       sync.Mutex
+	configs  []Config
 	clusters map[string]*Cluster
+	registry scraper.Registry
 	logger   *log.Logger
 }
 
-func NewService(c []Config, l *log.Logger) (*Service, error) {
+// NewService creates a new unopened k8s service
+func NewService(c []Config, r scraper.Registry, l *log.Logger) (*Service, error) {
 	l.Println("D! k8s configs", c)
 
 	clusters := make(map[string]*Cluster, len(c))
@@ -28,10 +33,13 @@ func NewService(c []Config, l *log.Logger) (*Service, error) {
 
 	return &Service{
 		clusters: clusters,
+		configs:  c,
 		logger:   l,
+		registry: r,
 	}, nil
 }
 
+// Open starts the kubernetes service
 func (s *Service) Open() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -40,26 +48,47 @@ func (s *Service) Open() error {
 			return err
 		}
 	}
-	return nil
+	s.register()
+	return s.registry.Commit()
 }
+
 func (s *Service) Close() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	for _, c := range s.clusters {
 		c.Close()
 	}
-	return nil
+	s.deregister()
+	return s.registry.Commit()
+}
+
+func (s *Service) deregister() {
+	// Remove all the configurations in the registry
+	for _, d := range s.configs {
+		s.registry.RemoveDiscoverer(&d)
+	}
+}
+
+func (s *Service) register() {
+	// Add all configurations to registry
+	for _, d := range s.configs {
+		if d.Enabled {
+			s.registry.AddDiscoverer(&d)
+		}
+	}
 }
 
 func (s *Service) Update(newConfigs []interface{}) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	configs := make([]Config, len(newConfigs))
 	existingClusters := make(map[string]bool, len(newConfigs))
 	for i := range newConfigs {
 		c, ok := newConfigs[i].(Config)
 		if !ok {
 			return fmt.Errorf("expected config object to be of type %T, got %T", c, newConfigs[i])
 		}
+		configs[i] = c
 		cluster, ok := s.clusters[c.ID]
 		if !ok {
 			var err error
@@ -86,6 +115,9 @@ func (s *Service) Update(newConfigs []interface{}) error {
 			delete(s.clusters, id)
 		}
 	}
+	s.deregister()
+	s.configs = configs
+	s.register()
 	return nil
 }
 

services/scraper/config.go

@@ -131,34 +131,3 @@ func decodeJobName(job string) (string, string, string, error) {
 	}
 	return split[0], split[1], split[2], nil
 }
-
-type KubernetesRole string
-
-const (
-	KubernetesRoleNode     = "node"
-	KubernetesRolePod      = "pod"
-	KubernetesRoleService  = "service"
-	KubernetesRoleEndpoint = "endpoints"
-)
-
-// Kubernetes is Kubernetes service discovery configuration
-type Kubernetes struct {
-	APIServer url.URL        `toml:"api_server" override:"api_server"`
-	Role      KubernetesRole `toml:"role" override:"role"`
-}
-
-func (k Kubernetes) Prom(c *config.ScrapeConfig) {
-	// TODO: auth token tls
-	c.ServiceDiscoveryConfig.KubernetesSDConfigs = []*config.KubernetesSDConfig{
-		&config.KubernetesSDConfig{
-			APIServer: config.URL{
-				URL: &k.APIServer,
-			},
-			Role: config.KubernetesRole(k.Role),
-		},
-	}
-}
-
-func NewKubernetes() Kubernetes {
-	return Kubernetes{}
-}