Repository: allegro/marathon-consul
Pull request #106 (2016-08-25 14:37:30)

Migrate to marathon task tag
Adds a script for migrating registered services to the state expected by #99, i.e. containing marathon-task tag with related marathon task ID.

This script is expected to be executed on the consul cluster with services managed by marathon-consul, in all DCs separately, before updating an existing marathon-consul instance. If consul services are not migrated before update, services managed by marathon-consul will get deregistered and registered again according to the new format during sync at first startup following the update. This will cause ~10 seconds service unavailability in the discovery before service healthchecks become passing.

janisz: Would you mind updating Readme please 

Patch:
diff --git a/.travis.yml b/.travis.yml
index 72295ba..6d950bb 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -19,6 +19,7 @@ script:
   - go test -coverprofile=metrics.coverprofile ./metrics
   - go test -coverprofile=service.coverprofile ./service
   - go test -coverprofile=sync.coverprofile ./sync
+  - go test -coverprofile=scripts.migrate_to_marathon_task_tag.coverprofile ./scripts/migrate_to_marathon_task_tag
   - go test -coverprofile=utils.coverprofile ./utils
   - go test -coverprofile=web.coverprofile ./web
   - go test -coverprofile=main.coverprofile
diff --git a/README.md b/README.md
index 52b340a..e4408bf 100644
--- a/README.md
+++ b/README.md
@@ -145,13 +145,15 @@ curl -X GET http://localhost:8500/v1/catalog/service/my-new-app
 "ServiceTags": [
   "marathon",
   "varnish",
-  "metrics"
+  "metrics",
+  "marathon-task:my-new-app.6a95bb03-6ad3-11e6-beaf-080027a7aca0"
 ],
 
 ```
+- Every service registration contains an additional tag `marathon-task` specifying the Marathon task id related to this registration.
 - If there are multiple ports in use for the same app, note that only the first one will be registered by marathon-consul in Consul.
 
-### Task heathchecks
+### Task healthchecks
 
 - At least one HTTP healthcheck should be defined for a task. The task is registered when Marathon marks it as alive.
 - The provided HTTP healthcheck will be transferred to Consul.
@@ -222,6 +224,30 @@ The following section describes known limitations in `marathon-consul`.
   Later on, if another deployment takes place, new services are registered with a new name, the old ones are not being deregistered though.
   A scheduled sync is required to wipe them out.
 
+## Migration to version 1.x.x
+
+Until 1.x.x marathon-consul would register services in Consul with registration id equal to related Marathon task id. Since 1.x.x registration ids are different and
+an additional tag, `marathon-task`, is added to each registration.
+
+If you update marathon-consul from version 0.x.x to 1.x.x, expect the synchronization phase during the first startup to reregister all services managed by marathon-consul to the new format.
+Depending on the configured healthchecks this may result in many seconds of services unavailability in the registry before their healthchecks become passing. If that's
+not acceptable for you, you should ensure that before updating marathon-consul all managed services contain a `marathon-task:<task-id>` tag, where `<task-id>` is the Marathon task id
+being registered in Consul.
+
+### Migration script
+
+You can run our migration script before updating to 1.x.x to ensure marathon-task tag in each registered service in a given datacenter.
+
+```
+migrate_to_marathon_task_tag --bootstrap-agent-location=1.2.3.4:8500
+```
+
+Provide `--bootstrap-agent-location` specifying address and port of one of the nodes in migrated datacenter. The script will discover
+all nodes in the datacenter and migrate all services registered there. Optionally, you can provide `--consul-tag` to match the one
+configured in your marathon-consul installation if you did customize it.
+
+If you have multiple datacenters you should migrate each one separately.
+
 ## License
 
 Marathon-consul is released under the Apache 2.0 license (see [LICENSE](LICENSE))
diff --git a/scripts/migrate_to_marathon_task_tag/main.go b/scripts/migrate_to_marathon_task_tag/main.go
new file mode 100644
index 0000000..a61b760
--- /dev/null
+++ b/scripts/migrate_to_marathon_task_tag/main.go
@@ -0,0 +1,162 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/hashicorp/consul/api"
+)
+
+type Config struct {
+	BootstrapAgentLocation string
+	ConsulTag              string
+}
+
+type MigrationStats struct {
+	MigratedServices                          int
+	SkippedFailedNodes                        int
+	SkippedFailedServices                     int
+	SkippedServicesNotManagedByMarathonConsul int
+	SkippedServicesAlreadyWithMarathonTaskTag int
+}
+
+func main() {
+	config := createConfig()
+	migrateSingleDC(config.BootstrapAgentLocation, config.ConsulTag)
+}
+
+func createConfig() *Config {
+	config := &Config{}
+	flag.StringVar(&config.BootstrapAgentLocation, "bootstrap-agent-location", "127.0.0.1:8500", "Address with port to one of the agents in the migrated DC")
+	flag.StringVar(&config.ConsulTag, "consul-tag", "marathon", "Tag specifying that a service is managed by marathon-consul")
+	flag.Parse()
+	return config
+}
+
+func migrateSingleDC(bootstrapAgentLocation string, consulTag string) *MigrationStats {
+	log.WithField("BootstrapAgentLocation", bootstrapAgentLocation).WithField("Tag", consulTag).Info("Starting migration...")
+
+	agentsPort, err := extractPort(bootstrapAgentLocation)
+	if err != nil {
+		log.WithError(err).WithField("BootstrapAgentLocation", bootstrapAgentLocation).
+			Fatal("Could not extract port from agent location, aborting.")
+		os.Exit(1)
+	}
+
+	client, err := api.NewClient(&api.Config{
+		Address: bootstrapAgentLocation,
+	})
+	if err != nil {
+		log.WithError(err).WithField("BootstrapAgentLocation", bootstrapAgentLocation).
+			Fatal("Could not create client to agent, aborting.")
+		os.Exit(1)
+	}
+
+	nodes, _, err := client.Catalog().Nodes(&api.QueryOptions{})
+	if err != nil {
+		log.WithError(err).Fatal("Could not fetch nodes, aborting.")
+		os.Exit(1)
+	}
+
+	log.Infof("Discovered %d node(s) to migrate", len(nodes))
+	stats := migrateNodes(nodes, agentsPort, consulTag)
+	log.WithField("Stats", stats).Info("Migration finished")
+	return stats
+}
+
+func migrateNodes(nodes []*api.Node, agentsPort int, consulTag string) *MigrationStats {
+	stats := &MigrationStats{}
+
+	for _, node := range nodes {
+		log.WithField("Node", node.Node).Info("Migrating node...")
+		nodeAddress := fmt.Sprintf("%s:%d", node.Address, agentsPort)
+		nodeClient, err := api.NewClient(&api.Config{
+			Address: nodeAddress,
+		})
+		if err != nil {
+			log.WithError(err).WithField("Node", node.Node).WithField("Address", nodeAddress).
+				Warn("Could not create client to node, skipping")
+			stats.SkippedFailedNodes++
+			continue
+		}
+		agentServices, err := nodeClient.Agent().Services()
+		if err != nil {
+			log.WithError(err).WithField("Node", node.Node).Warn("Could not fetch services on node, skipping")
+			stats.SkippedFailedNodes++
+			continue
+		}
+
+		migrateServicesOnNode(agentServices, nodeClient, consulTag, stats)
+		log.WithField("Node", node.Node).Info("Migrated node")
+	}
+
+	return stats
+}
+
+func migrateServicesOnNode(services map[string]*api.AgentService, nodeClient *api.Client, consulTag string, stats *MigrationStats) {
+	for _, agentService := range services {
+		if !isManagedByMarathonConsul(agentService, consulTag) {
+			log.WithField("ServiceID", agentService.ID).Info("Service not managed by marathon-consul, skipping")
+			stats.SkippedServicesNotManagedByMarathonConsul++
+			continue
+		}
+		if hasMarathonTaskTag(agentService) {
+			log.WithField("ServiceID", agentService.ID).Info("Service already has marathon-task tag, skipping")
+			stats.SkippedServicesAlreadyWithMarathonTaskTag++
+			continue
+		}
+
+		if err := nodeClient.Agent().ServiceRegister(migrated(agentService)); err != nil {
+			log.WithError(err).WithField("ServiceID", agentService.ID).
+			Warn("Could not reregister service, skipping")
+			stats.SkippedFailedServices++
+			continue
+		}
+
+		log.WithField("ServiceID", agentService.ID).Info("Migrated service")
+		stats.MigratedServices++
+	}
+}
+
+func extractPort(address string) (int, error) {
+	indexOfColon := strings.LastIndex(address, ":")
+	if indexOfColon == -1 {
+		return -1, fmt.Errorf("Could not extract port from address %s, colon not found", address)
+	}
+	return strconv.Atoi(address[indexOfColon+1:])
+}
+
+func isManagedByMarathonConsul(checked *api.AgentService, consulTag string) bool {
+	for _, tag := range checked.Tags {
+		if tag == consulTag {
+			return true
+		}
+	}
+	return false
+}
+
+func hasMarathonTaskTag(checked *api.AgentService) bool {
+	for _, tag := range checked.Tags {
+		if strings.HasPrefix(tag, "marathon-task:") {
+			return true
+		}
+	}
+	return false
+}
+
+func migrated(toMigrate *api.AgentService) *api.AgentServiceRegistration {
+	tags := append(toMigrate.Tags, fmt.Sprintf("marathon-task:%s", toMigrate.ID))
+
+	return &api.AgentServiceRegistration{
+		ID:                toMigrate.ID,
+		Name:              toMigrate.Service,
+		Tags:              tags,
+		Port:              toMigrate.Port,
+		Address:           toMigrate.Address,
+		EnableTagOverride: toMigrate.EnableTagOverride,
+	}
+}
diff --git a/sync/sync.go b/sync/sync.go
index a0e8ffd..57f7746 100644
--- a/sync/sync.go
+++ b/sync/sync.go
@@ -122,12 +122,13 @@ func (s *Sync) deregisterConsulServicesNotFoundInMarathon(marathonApps []*apps.A
 	runningTasks := s.marathonTaskIdsSet(marathonApps)
 	for _, service := range services {
 		taskIdInTag, err := service.TaskId()
-		if err != nil {
-			log.WithField("Id", service.ID).WithError(err).Warn("Couldn't extract marathon task id, skipping sync for this service")
-			continue
+		taskIdNotFoundInTag := err != nil
+		if taskIdNotFoundInTag {
+			log.WithField("Id", service.ID).WithError(err).
+				Warn("Couldn't extract marathon task id, deregistering to have sync reregister it properly")
 		}
 
-		if _, isRunning := runningTasks[taskIdInTag]; !isRunning {
+		if _, isRunning := runningTasks[taskIdInTag]; !isRunning || taskIdNotFoundInTag {
 			err := s.serviceRegistry.Deregister(service)
 			if err != nil {
 				log.WithError(err).WithFields(log.Fields{


Test patch:
diff --git a/scripts/migrate_to_marathon_task_tag/main_test.go b/scripts/migrate_to_marathon_task_tag/main_test.go
new file mode 100644
index 0000000..3071d7d
--- /dev/null
+++ b/scripts/migrate_to_marathon_task_tag/main_test.go
@@ -0,0 +1,127 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/allegro/marathon-consul/consul"
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/testutil"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMigrateToMarathonTaskTag_shouldMigrateOnlyServicesWithMarathonTagAndWithoutTaskTag(t *testing.T) {
+	t.Parallel()
+	// given
+	server := consul.CreateConsulTestServer(t)
+	defer server.Stop()
+	server.AddService("serviceA", "passing", []string{"marathon"})
+	server.AddService("serviceB-critical", "critical", []string{"marathon"})
+	server.AddService("serviceC-with-marathon-task-tag", "passing", []string{"marathon", "marathon-task:some-task"})
+	server.AddService("serviceD-without-marathon-tag", "passing", []string{})
+
+	// when
+	stats := migrateSingleDC(server.HTTPAddr, "marathon")
+
+	// then
+	client, _ := clientToServer(server)
+	services, _, _ := client.Catalog().Services(&api.QueryOptions{})
+	assert.Contains(t, services["serviceA"], "marathon-task:serviceA")
+	assert.Contains(t, services["serviceB-critical"], "marathon-task:serviceB-critical")
+	assert.NotContains(t, services["serviceC-with-marathon-task-tag"], "marathon-task:serviceC-with-marathon-task-tag")
+	assert.NotContains(t, services["serviceD-without-marathon-tag"], "marathon-task:serviceD-without-marathon-tag")
+	assert.Equal(t, 2, stats.MigratedServices)
+	assert.Equal(t, 2, stats.SkippedServicesNotManagedByMarathonConsul) // +1 for service consul
+	assert.Equal(t, 1, stats.SkippedServicesAlreadyWithMarathonTaskTag)
+}
+
+func TestMigrateToMarathonTaskTag_shouldLeaveServicePropertiesOtherThanTagsUnchanged(t *testing.T) {
+	t.Parallel()
+	// given
+	server := consul.CreateConsulTestServer(t)
+	defer server.Stop()
+	client, _ := clientToServer(server)
+	client.Agent().ServiceRegister(&api.AgentServiceRegistration{
+		ID:      "serviceA.0",
+		Name:    "serviceA",
+		Tags:    []string{"marathon"},
+		Port:    1337,
+		Address: "1.2.3.4",
+		Check: &api.AgentServiceCheck{
+			HTTP:     "http://1.2.3.4/health",
+			Interval: "60s",
+		},
+	})
+	entriesBeforeMigration, _, _ := client.Health().Service("serviceA", "", false, &api.QueryOptions{})
+	before := entriesBeforeMigration[0]
+
+	// when
+	migrateSingleDC(server.HTTPAddr, "marathon")
+
+	// then
+	entriesAfterMigration, _, _ := client.Health().Service("serviceA", "", false, &api.QueryOptions{})
+	entryAfterMigrationAsJSON, _ := json.Marshal(entriesAfterMigration[0])
+	expectedEntryAfterMigrationAsJSON, _ := json.Marshal(&api.ServiceEntry{
+		Node:   before.Node,
+		Checks: before.Checks,
+		Service: &api.AgentService{
+			ID:                before.Service.ID,
+			Service:           before.Service.Service,
+			Tags:              []string{"marathon", fmt.Sprintf("marathon-task:%s", before.Service.ID)},
+			Port:              before.Service.Port,
+			Address:           before.Service.Address,
+			EnableTagOverride: before.Service.EnableTagOverride,
+		},
+	})
+	assert.Equal(t, string(expectedEntryAfterMigrationAsJSON), string(entryAfterMigrationAsJSON))
+}
+
+func TestMigrateToMarathonTaskTag_shouldMigrateServicesInBootstrapAgentsDCOnly(t *testing.T) {
+	t.Parallel()
+	// given
+	server := consul.CreateConsulTestServer(t)
+	defer server.Stop()
+	serverOnOtherDC := consul.CreateConsulTestServer(t)
+	defer serverOnOtherDC.Stop()
+	server.JoinWAN(serverOnOtherDC.WANAddr)
+
+	server.AddService("serviceA", "passing", []string{"marathon"})
+	serverOnOtherDC.AddService("serviceB", "passing", []string{"marathon"})
+
+	// when
+	migrateSingleDC(server.HTTPAddr, "marathon")
+
+	// then
+	client, _ := clientToServer(server)
+	clientToOtherDC, _ := clientToServer(serverOnOtherDC)
+	services, _, _ := client.Catalog().Services(&api.QueryOptions{})
+	servicesOnOtherDC, _, _ := clientToOtherDC.Catalog().Services(&api.QueryOptions{})
+	assert.Contains(t, services["serviceA"], "marathon-task:serviceA")
+	assert.NotContains(t, servicesOnOtherDC["serviceB"], "marathon-task:serviceC")
+}
+
+func TestMigrateToMarathonTaskTag_shouldMigrateThroughCallingMain(t *testing.T) {
+	t.Parallel()
+	// given
+	server := consul.CreateConsulTestServer(t)
+	defer server.Stop()
+	server.AddService("serviceA", "passing", []string{"marathon"})
+	os.Args = []string{"./migrate", fmt.Sprintf("--bootstrap-agent-location=%s", server.HTTPAddr)}
+
+	// when
+	main()
+
+	//then
+	client, _ := clientToServer(server)
+	services, _, _ := client.Catalog().Services(&api.QueryOptions{})
+	assert.Contains(t, services["serviceA"], "marathon-task:serviceA")
+}
+
+func clientToServer(server *testutil.TestServer) (*api.Client, error) {
+	return api.NewClient(&api.Config{
+		Address:    server.HTTPAddr,
+		Datacenter: server.Config.Datacenter,
+	})
+}
diff --git a/sync/sync_test.go b/sync/sync_test.go
index b674c43..3c87cdb 100644
--- a/sync/sync_test.go
+++ b/sync/sync_test.go
@@ -295,7 +295,7 @@ func TestSyncOnlyHealthyServices(t *testing.T) {
 	}
 }
 
-func TestSync_SkipServicesWithoutMarathonTaskTag(t *testing.T) {
+func TestSync_DeregisterServicesWithoutMarathonTaskTag(t *testing.T) {
 	t.Parallel()
 	// given
 	app := ConsulApp("app1", 1)
@@ -308,8 +308,7 @@ func TestSync_SkipServicesWithoutMarathonTaskTag(t *testing.T) {
 
 	// then
 	services, _ := consul.GetAllServices()
-	assert.Equal(t, 1, len(services))
-	assert.NotContains(t, services[0].Tags, service.MarathonTaskTag(app.Tasks[0].ID))
+	assert.Empty(t, services)
 }
 
 func TestSync_WithRegisteringProblems(t *testing.T) {