provider/cluster: ignore stale, un-owned leases. (#382)

fixes #380

Author: boz

provider/cluster/service.go

@@ -50,16 +50,11 @@ func NewService(ctx context.Context, session session.Session, bus event.Bus, cli
 		return nil, err
 	}
 
-	deployments, err := client.Deployments()
+	deployments, err := findDeployments(ctx, log, client, session)
 	if err != nil {
-		log.Error("fetching deployments", "err", err)
 		sub.Close()
 		return nil, err
 	}
-	log.Info("found managed deployments", "count", len(deployments))
-	for _, deployment := range deployments {
-		log.Debug("deployment", "lease", deployment.LeaseID(), "mgroup", deployment.ManifestGroup().Name)
-	}
 
 	inventory, err := newInventoryService(config, log, lc.ShuttingDown(), sub, client, deployments)
 	if err != nil {
@@ -262,3 +257,45 @@ func (s *service) teardownLease(lid types.LeaseID) {
 		s.log.Error("tearing down lease deployment", "err", err, "lease", lid)
 	}
 }
+
+func findDeployments(ctx context.Context, log log.Logger, client Client, session session.Session) ([]Deployment, error) {
+	deployments, err := client.Deployments()
+	if err != nil {
+		log.Error("fetching deployments", "err", err)
+		return nil, err
+	}
+
+	leaseList, err := session.Query().Leases(ctx)
+	if err != nil {
+		log.Error("fetching deployments", "err", err)
+		return nil, err
+	}
+
+	leases := make(map[string]*types.Lease, len(leaseList.Items))
+	for _, lease := range leaseList.Items {
+		if !lease.Provider.Equal(session.Provider().Address) {
+			continue
+		}
+		if lease.State != types.Lease_ACTIVE {
+			continue
+		}
+		leases[lease.Path()] = lease
+	}
+
+	log.Info("found leases", "num-active", len(leases), "num-skipped", len(leaseList.Items)-len(leases))
+
+	dcount := len(deployments)
+	for idx, deployment := range deployments {
+		if _, ok := leases[deployment.LeaseID().Path()]; !ok {
+			continue
+		}
+
+		deployments = append(deployments[:idx], deployments[idx+1:]...)
+
+		log.Debug("deployment", "lease", deployment.LeaseID(), "mgroup", deployment.ManifestGroup().Name)
+	}
+
+	log.Info("found deployments", "num-active", len(deployments), "num-skipped", dcount-len(deployments))
+
+	return deployments, nil
+}

provider/cluster/service_test.go

@@ -117,7 +117,7 @@ func withServiceTestSetup(t *testing.T, fn func(event.Bus, types.LeaseID)) {
 		Return(cluster.NullClient().Inventory()).
 		Maybe()
 
-	c, err := cluster.NewService(ctx, providerSession(t), bus, client)
+	c, err := cluster.NewService(ctx, providerSession(t, lease), bus, client)
 	require.NoError(t, err)
 	testutil.WaitReady(t, c.Ready())
 
@@ -143,10 +143,14 @@ func withServiceTestSetup(t *testing.T, fn func(event.Bus, types.LeaseID)) {
 	mock.AssertExpectationsForObjects(t, client)
 }
 
-func providerSession(t *testing.T) session.Session {
+func providerSession(t *testing.T, leases ...*types.Lease) session.Session {
 	log := testutil.Logger()
 	txc := new(txumocks.Client)
 	qc := new(qmocks.Client)
+
+	qc.On("Leases", mock.Anything).
+		Return(&types.Leases{Items: leases}, nil)
+
 	provider := testutil.Provider(testutil.Address(t), 1)
 	return session.New(log, provider, txc, qc)
 }