Merge pull request #1590 from prometheus/fabxc-marathon-ref

Fix basic issues in marathon SD

retrieval/discovery/marathon.go

@@ -16,81 +16,15 @@ package discovery
 import (
 	"time"
 
-	"github.com/prometheus/common/log"
-	"golang.org/x/net/context"
-
 	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/retrieval/discovery/marathon"
 )
 
-// MarathonDiscovery provides service discovery based on a Marathon instance.
+// NewMarathon creates a new Marathon based discovery.
-type MarathonDiscovery struct {
+func NewMarathon(conf *config.MarathonSDConfig) *marathon.Discovery {
-	servers         []string
+	return &marathon.Discovery{
-	refreshInterval time.Duration
+		Servers:         conf.Servers,
-	done            chan struct{}
+		RefreshInterval: time.Duration(conf.RefreshInterval),
-	lastRefresh     map[string]*config.TargetGroup
+		Client:          marathon.FetchApps,
-	client          marathon.AppListClient
-}
-
-// NewMarathonDiscovery creates a new Marathon based discovery.
-func NewMarathonDiscovery(conf *config.MarathonSDConfig) *MarathonDiscovery {
-	return &MarathonDiscovery{
-		servers:         conf.Servers,
-		refreshInterval: time.Duration(conf.RefreshInterval),
-		done:            make(chan struct{}),
-		client:          marathon.FetchMarathonApps,
 	}
 }
-
-// Run implements the TargetProvider interface.
-func (md *MarathonDiscovery) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
-	defer close(ch)
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-time.After(md.refreshInterval):
-			err := md.updateServices(ch)
-			if err != nil {
-				log.Errorf("Error while updating services: %s", err)
-			}
-		}
-	}
-}
-
-func (md *MarathonDiscovery) updateServices(ch chan<- []*config.TargetGroup) error {
-	targetMap, err := md.fetchTargetGroups()
-	if err != nil {
-		return err
-	}
-
-	all := make([]*config.TargetGroup, 0, len(targetMap))
-	for _, tg := range targetMap {
-		all = append(all, tg)
-	}
-	ch <- all
-
-	// Remove services which did disappear
-	for source := range md.lastRefresh {
-		_, ok := targetMap[source]
-		if !ok {
-			log.Debugf("Removing group for %s", source)
-			ch <- []*config.TargetGroup{{Source: source}}
-		}
-	}
-
-	md.lastRefresh = targetMap
-	return nil
-}
-
-func (md *MarathonDiscovery) fetchTargetGroups() (map[string]*config.TargetGroup, error) {
-	url := marathon.RandomAppsURL(md.servers)
-	apps, err := md.client(url)
-	if err != nil {
-		return nil, err
-	}
-
-	groups := marathon.AppsToTargetGroups(apps)
-	return groups, nil
-}

retrieval/discovery/marathon/client.go

@@ -1,47 +0,0 @@
-// Copyright 2015 The Prometheus Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package marathon
-
-import (
-	"encoding/json"
-	"io/ioutil"
-	"net/http"
-)
-
-// AppListClient defines a function that can be used to get an application list from marathon.
-type AppListClient func(url string) (*AppList, error)
-
-// FetchMarathonApps requests a list of applications from a marathon server.
-func FetchMarathonApps(url string) (*AppList, error) {
-	resp, err := http.Get(url)
-	if err != nil {
-		return nil, err
-	}
-
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return nil, err
-	}
-
-	return parseAppJSON(body)
-}
-
-func parseAppJSON(body []byte) (*AppList, error) {
-	apps := &AppList{}
-	err := json.Unmarshal(body, apps)
-	if err != nil {
-		return nil, err
-	}
-	return apps, nil
-}

retrieval/discovery/marathon/constants.go

@@ -1,32 +0,0 @@
-// Copyright 2015 The Prometheus Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package marathon
-
-import (
-	"github.com/prometheus/common/model"
-)
-
-const (
-	// metaLabelPrefix is the meta prefix used for all meta labels in this discovery.
-	metaLabelPrefix = model.MetaLabelPrefix + "marathon_"
-	// appLabelPrefix is the prefix for the application labels.
-	appLabelPrefix = metaLabelPrefix + "app_label_"
-
-	// appLabel is used for the name of the app in Marathon.
-	appLabel model.LabelName = metaLabelPrefix + "app"
-	// imageLabel is the label that is used for the docker image running the service.
-	imageLabel model.LabelName = metaLabelPrefix + "image"
-	// taskLabel contains the mesos task name of the app instance.
-	taskLabel model.LabelName = metaLabelPrefix + "task"
-)

retrieval/discovery/marathon/conversion.go

@@ -1,71 +0,0 @@
-// Copyright 2015 The Prometheus Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package marathon
-
-import (
-	"fmt"
-
-	"github.com/prometheus/common/model"
-
-	"github.com/prometheus/prometheus/config"
-)
-
-// AppsToTargetGroups takes an array of Marathon apps and converts them into target groups.
-func AppsToTargetGroups(apps *AppList) map[string]*config.TargetGroup {
-	tgroups := map[string]*config.TargetGroup{}
-	for _, a := range apps.Apps {
-		group := createTargetGroup(&a)
-		tgroups[group.Source] = group
-	}
-	return tgroups
-}
-
-func createTargetGroup(app *App) *config.TargetGroup {
-	var (
-		targets = targetsForApp(app)
-		appName = model.LabelValue(app.ID)
-		image   = model.LabelValue(app.Container.Docker.Image)
-	)
-	tg := &config.TargetGroup{
-		Targets: targets,
-		Labels: model.LabelSet{
-			appLabel:   appName,
-			imageLabel: image,
-		},
-		Source: app.ID,
-	}
-
-	for ln, lv := range app.Labels {
-		ln = appLabelPrefix + ln
-		tg.Labels[model.LabelName(ln)] = model.LabelValue(lv)
-	}
-
-	return tg
-}
-
-func targetsForApp(app *App) []model.LabelSet {
-	targets := make([]model.LabelSet, 0, len(app.Tasks))
-	for _, t := range app.Tasks {
-		target := targetForTask(&t)
-		targets = append(targets, model.LabelSet{
-			model.AddressLabel: model.LabelValue(target),
-			taskLabel:          model.LabelValue(t.ID),
-		})
-	}
-	return targets
-}
-
-func targetForTask(task *Task) string {
-	return fmt.Sprintf("%s:%d", task.Host, task.Ports[0])
-}

retrieval/discovery/marathon/marathon.go

@@ -0,0 +1,230 @@
+// Copyright 2016 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package marathon
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"net/http"
+	"time"
+
+	"golang.org/x/net/context"
+
+	"github.com/prometheus/common/log"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/config"
+)
+
+const (
+	// metaLabelPrefix is the meta prefix used for all meta labels in this discovery.
+	metaLabelPrefix = model.MetaLabelPrefix + "marathon_"
+	// appLabelPrefix is the prefix for the application labels.
+	appLabelPrefix = metaLabelPrefix + "app_label_"
+
+	// appLabel is used for the name of the app in Marathon.
+	appLabel model.LabelName = metaLabelPrefix + "app"
+	// imageLabel is the label that is used for the docker image running the service.
+	imageLabel model.LabelName = metaLabelPrefix + "image"
+	// taskLabel contains the mesos task name of the app instance.
+	taskLabel model.LabelName = metaLabelPrefix + "task"
+)
+
+const appListPath string = "/v2/apps/?embed=apps.tasks"
+
+// Discovery provides service discovery based on a Marathon instance.
+type Discovery struct {
+	Servers         []string
+	RefreshInterval time.Duration
+	lastRefresh     map[string]*config.TargetGroup
+	Client          AppListClient
+}
+
+// Run implements the TargetProvider interface.
+func (md *Discovery) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
+	defer close(ch)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(md.RefreshInterval):
+			err := md.updateServices(ctx, ch)
+			if err != nil {
+				log.Errorf("Error while updating services: %s", err)
+			}
+		}
+	}
+}
+
+func (md *Discovery) updateServices(ctx context.Context, ch chan<- []*config.TargetGroup) error {
+	targetMap, err := md.fetchTargetGroups()
+	if err != nil {
+		return err
+	}
+
+	all := make([]*config.TargetGroup, 0, len(targetMap))
+	for _, tg := range targetMap {
+		all = append(all, tg)
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case ch <- all:
+	}
+
+	// Remove services which did disappear.
+	for source := range md.lastRefresh {
+		_, ok := targetMap[source]
+		if !ok {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case ch <- []*config.TargetGroup{{Source: source}}:
+				log.Debugf("Removing group for %s", source)
+			}
+		}
+	}
+
+	md.lastRefresh = targetMap
+	return nil
+}
+
+func (md *Discovery) fetchTargetGroups() (map[string]*config.TargetGroup, error) {
+	url := RandomAppsURL(md.Servers)
+	apps, err := md.Client(url)
+	if err != nil {
+		return nil, err
+	}
+
+	groups := AppsToTargetGroups(apps)
+	return groups, nil
+}
+
+// Task describes one instance of a service running on Marathon.
+type Task struct {
+	ID    string   `json:"id"`
+	Host  string   `json:"host"`
+	Ports []uint32 `json:"ports"`
+}
+
+// DockerContainer describes a container which uses the docker runtime.
+type DockerContainer struct {
+	Image string `json:"image"`
+}
+
+// Container describes the runtime an app in running in.
+type Container struct {
+	Docker DockerContainer `json:"docker"`
+}
+
+// App describes a service running on Marathon.
+type App struct {
+	ID           string            `json:"id"`
+	Tasks        []Task            `json:"tasks"`
+	RunningTasks int               `json:"tasksRunning"`
+	Labels       map[string]string `json:"labels"`
+	Container    Container         `json:"container"`
+}
+
+// AppList is a list of Marathon apps.
+type AppList struct {
+	Apps []App `json:"apps"`
+}
+
+// AppListClient defines a function that can be used to get an application list from marathon.
+type AppListClient func(url string) (*AppList, error)
+
+// FetchApps requests a list of applications from a marathon server.
+func FetchApps(url string) (*AppList, error) {
+	resp, err := http.Get(url)
+	if err != nil {
+		return nil, err
+	}
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	return parseAppJSON(body)
+}
+
+func parseAppJSON(body []byte) (*AppList, error) {
+	apps := &AppList{}
+	err := json.Unmarshal(body, apps)
+	if err != nil {
+		return nil, err
+	}
+	return apps, nil
+}
+
+// RandomAppsURL randomly selects a server from an array and creates
+// an URL pointing to the app list.
+func RandomAppsURL(servers []string) string {
+	// TODO: If possible update server list from Marathon at some point.
+	server := servers[rand.Intn(len(servers))]
+	return fmt.Sprintf("%s%s", server, appListPath)
+}
+
+// AppsToTargetGroups takes an array of Marathon apps and converts them into target groups.
+func AppsToTargetGroups(apps *AppList) map[string]*config.TargetGroup {
+	tgroups := map[string]*config.TargetGroup{}
+	for _, a := range apps.Apps {
+		group := createTargetGroup(&a)
+		tgroups[group.Source] = group
+	}
+	return tgroups
+}
+
+func createTargetGroup(app *App) *config.TargetGroup {
+	var (
+		targets = targetsForApp(app)
+		appName = model.LabelValue(app.ID)
+		image   = model.LabelValue(app.Container.Docker.Image)
+	)
+	tg := &config.TargetGroup{
+		Targets: targets,
+		Labels: model.LabelSet{
+			appLabel:   appName,
+			imageLabel: image,
+		},
+		Source: app.ID,
+	}
+
+	for ln, lv := range app.Labels {
+		ln = appLabelPrefix + ln
+		tg.Labels[model.LabelName(ln)] = model.LabelValue(lv)
+	}
+
+	return tg
+}
+
+func targetsForApp(app *App) []model.LabelSet {
+	targets := make([]model.LabelSet, 0, len(app.Tasks))
+	for _, t := range app.Tasks {
+		target := targetForTask(&t)
+		targets = append(targets, model.LabelSet{
+			model.AddressLabel: model.LabelValue(target),
+			taskLabel:          model.LabelValue(t.ID),
+		})
+	}
+	return targets
+}
+
+func targetForTask(task *Task) string {
+	return fmt.Sprintf("%s:%d", task.Host, task.Ports[0])
+}

retrieval/discovery/marathon/marathon_test.go

@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package discovery
+package marathon
 
 import (
 	"errors"
@@ -22,23 +22,22 @@ import (
 	"golang.org/x/net/context"
 
 	"github.com/prometheus/prometheus/config"
-	"github.com/prometheus/prometheus/retrieval/discovery/marathon"
 )
 
 var marathonValidLabel = map[string]string{"prometheus": "yes"}
 
-func newTestDiscovery(client marathon.AppListClient) (chan []*config.TargetGroup, *MarathonDiscovery) {
+func newTestDiscovery(client AppListClient) (chan []*config.TargetGroup, *Discovery) {
 	ch := make(chan []*config.TargetGroup)
-	md := NewMarathonDiscovery(&config.MarathonSDConfig{
+	md := &Discovery{
 		Servers: []string{"http://localhost:8080"},
-	})
+		Client:  client,
-	md.client = client
+	}
 	return ch, md
 }
 
 func TestMarathonSDHandleError(t *testing.T) {
 	var errTesting = errors.New("testing failure")
-	ch, md := newTestDiscovery(func(url string) (*marathon.AppList, error) {
+	ch, md := newTestDiscovery(func(url string) (*AppList, error) {
 		return nil, errTesting
 	})
 	go func() {
@@ -48,15 +47,15 @@ func TestMarathonSDHandleError(t *testing.T) {
 		default:
 		}
 	}()
-	err := md.updateServices(ch)
+	err := md.updateServices(context.Background(), ch)
 	if err != errTesting {
 		t.Fatalf("Expected error: %s", err)
 	}
 }
 
 func TestMarathonSDEmptyList(t *testing.T) {
-	ch, md := newTestDiscovery(func(url string) (*marathon.AppList, error) {
+	ch, md := newTestDiscovery(func(url string) (*AppList, error) {
-		return &marathon.AppList{}, nil
+		return &AppList{}, nil
 	})
 	go func() {
 		select {
@@ -67,34 +66,34 @@ func TestMarathonSDEmptyList(t *testing.T) {
 		default:
 		}
 	}()
-	err := md.updateServices(ch)
+	err := md.updateServices(context.Background(), ch)
 	if err != nil {
 		t.Fatalf("Got error: %s", err)
 	}
 }
 
-func marathonTestAppList(labels map[string]string, runningTasks int) *marathon.AppList {
+func marathonTestAppList(labels map[string]string, runningTasks int) *AppList {
-	task := marathon.Task{
+	task := Task{
 		ID:    "test-task-1",
 		Host:  "mesos-slave1",
 		Ports: []uint32{31000},
 	}
-	docker := marathon.DockerContainer{Image: "repo/image:tag"}
+	docker := DockerContainer{Image: "repo/image:tag"}
-	container := marathon.Container{Docker: docker}
+	container := Container{Docker: docker}
-	app := marathon.App{
+	app := App{
 		ID:           "test-service",
-		Tasks:        []marathon.Task{task},
+		Tasks:        []Task{task},
 		RunningTasks: runningTasks,
 		Labels:       labels,
 		Container:    container,
 	}
-	return &marathon.AppList{
+	return &AppList{
-		Apps: []marathon.App{app},
+		Apps: []App{app},
 	}
 }
 
 func TestMarathonSDSendGroup(t *testing.T) {
-	ch, md := newTestDiscovery(func(url string) (*marathon.AppList, error) {
+	ch, md := newTestDiscovery(func(url string) (*AppList, error) {
 		return marathonTestAppList(marathonValidLabel, 1), nil
 	})
 	go func() {
@@ -116,14 +115,14 @@ func TestMarathonSDSendGroup(t *testing.T) {
 			t.Fatal("Did not get a target group.")
 		}
 	}()
-	err := md.updateServices(ch)
+	err := md.updateServices(context.Background(), ch)
 	if err != nil {
 		t.Fatalf("Got error: %s", err)
 	}
 }
 
 func TestMarathonSDRemoveApp(t *testing.T) {
-	ch, md := newTestDiscovery(func(url string) (*marathon.AppList, error) {
+	ch, md := newTestDiscovery(func(url string) (*AppList, error) {
 		return marathonTestAppList(marathonValidLabel, 1), nil
 	})
 
@@ -137,32 +136,32 @@ func TestMarathonSDRemoveApp(t *testing.T) {
 			}
 		}
 	}()
-	err := md.updateServices(ch)
+	err := md.updateServices(context.Background(), ch)
 	if err != nil {
 		t.Fatalf("Got error on first update: %s", err)
 	}
 
-	md.client = func(url string) (*marathon.AppList, error) {
+	md.Client = func(url string) (*AppList, error) {
 		return marathonTestAppList(marathonValidLabel, 0), nil
 	}
-	err = md.updateServices(ch)
+	err = md.updateServices(context.Background(), ch)
 	if err != nil {
 		t.Fatalf("Got error on second update: %s", err)
 	}
 }
 
 func TestMarathonSDRunAndStop(t *testing.T) {
-	ch, md := newTestDiscovery(func(url string) (*marathon.AppList, error) {
+	ch, md := newTestDiscovery(func(url string) (*AppList, error) {
 		return marathonTestAppList(marathonValidLabel, 1), nil
 	})
-	md.refreshInterval = time.Millisecond * 10
+	md.RefreshInterval = time.Millisecond * 10
 	ctx, cancel := context.WithCancel(context.Background())
 
 	go func() {
 		select {
 		case <-ch:
 			cancel()
-		case <-time.After(md.refreshInterval * 3):
+		case <-time.After(md.RefreshInterval * 3):
 			cancel()
 			t.Fatalf("Update took too long.")
 		}

retrieval/discovery/marathon/objects.go

@@ -1,45 +0,0 @@
-// Copyright 2015 The Prometheus Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package marathon
-
-// Task describes one instance of a service running on Marathon.
-type Task struct {
-	ID    string   `json:"id"`
-	Host  string   `json:"host"`
-	Ports []uint32 `json:"ports"`
-}
-
-// DockerContainer describes a container which uses the docker runtime.
-type DockerContainer struct {
-	Image string `json:"image"`
-}
-
-// Container describes the runtime an app in running in.
-type Container struct {
-	Docker DockerContainer `json:"docker"`
-}
-
-// App describes a service running on Marathon.
-type App struct {
-	ID           string            `json:"id"`
-	Tasks        []Task            `json:"tasks"`
-	RunningTasks int               `json:"tasksRunning"`
-	Labels       map[string]string `json:"labels"`
-	Container    Container         `json:"container"`
-}
-
-// AppList is a list of Marathon apps.
-type AppList struct {
-	Apps []App `json:"apps"`
-}

retrieval/discovery/marathon/url.go

@@ -1,29 +0,0 @@
-// Copyright 2015 The Prometheus Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package marathon
-
-import (
-	"fmt"
-	"math/rand"
-)
-
-const appListPath string = "/v2/apps/?embed=apps.tasks"
-
-// RandomAppsURL randomly selects a server from an array and creates
-// an URL pointing to the app list.
-func RandomAppsURL(servers []string) string {
-	// TODO: If possible update server list from Marathon at some point.
-	server := servers[rand.Intn(len(servers))]
-	return fmt.Sprintf("%s%s", server, appListPath)
-}

retrieval/targetmanager.go

@@ -374,7 +374,7 @@ func providersFromConfig(cfg *config.ScrapeConfig) map[string]TargetProvider {
 		app("consul", i, k)
 	}
 	for i, c := range cfg.MarathonSDConfigs {
-		app("marathon", i, discovery.NewMarathonDiscovery(c))
+		app("marathon", i, discovery.NewMarathon(c))
 	}
 	for i, c := range cfg.KubernetesSDConfigs {
 		k, err := discovery.NewKubernetesDiscovery(c)