Fix service registry clustering

broker/big_poll_apply_manifest.go

@@ -0,0 +1,44 @@
+package broker
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"code.cloudfoundry.org/cli/api/cloudcontroller/ccv3"
+)
+
+func (broker *SCSBroker) PollApplyManifest(jobURL ccv3.JobURL) (ccv3.Warnings, error) {
+	var allWarnings ccv3.Warnings
+	cfClient, err := broker.GetClient()
+	if err != nil {
+		return nil, errors.New("Couldn't start session: " + err.Error())
+	}
+	count := 0
+
+	for {
+		time.Sleep(time.Second)
+		count += 1
+
+		broker.Logger.Info(fmt.Sprintf("Polling iteration %d, job %s", count, jobURL))
+		job, warnings, err := cfClient.GetJob(jobURL)
+		allWarnings = append(allWarnings, warnings...)
+		broker.showWarnings(warnings, "poll-apply-manifest")
+		if err != nil {
+			return allWarnings, err
+		}
+
+		broker.Logger.Info(fmt.Sprintf("HERE'S THE FUCKING JOB STATE: %s", job.State))
+		if job.HasFailed() {
+			err = job.Errors()[0]
+			broker.logWorkflowError("pollApplyManifest", "*none*", err)
+			return allWarnings, err
+		}
+
+		if job.IsComplete() {
+			return allWarnings, nil
+		}
+	}
+
+	return allWarnings, nil
+}

broker/bind.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	"github.com/cloudfoundry-community/go-uaa"
-	brokerapi "github.com/pivotal-cf/brokerapi/domain"
+	brokerapi "github.com/pivotal-cf/brokerapi/v7/domain"
 	"github.com/starkandwayne/scs-broker/broker/utilities"
 )
 

broker/broker.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	"code.cloudfoundry.org/lager"
-	brokerapi "github.com/pivotal-cf/brokerapi/domain"
+	brokerapi "github.com/pivotal-cf/brokerapi/v7/domain"
 	"github.com/starkandwayne/scs-broker/config"
 )
 

broker/client.go

@@ -3,6 +3,7 @@ package broker
 import (
 	"code.cloudfoundry.org/cli/api/cloudcontroller/ccv3"
 	clients "github.com/cloudfoundry-community/go-cf-clients-helper"
+	cf "github.com/cloudfoundry-community/go-cfclient"
 	"github.com/cloudfoundry-community/go-uaa"
 )
 
@@ -22,6 +23,17 @@ func (broker *SCSBroker) GetClient() (*ccv3.Client, error) {
 	return session.V3(), err
 }
 
+func (broker *SCSBroker) GetCommunity() (*cf.Client, error) {
+	config := &cf.Config{
+		ApiAddress:        broker.Config.CfConfig.ApiUrl,
+		SkipSslValidation: broker.Config.CfConfig.SkipSslValidation,
+		Username:          broker.Config.CfConfig.CfUsername,
+		Password:          broker.Config.CfConfig.CfPassword,
+	}
+
+	return cf.NewClient(config)
+}
+
 func (broker *SCSBroker) GetUaaClient() (*uaa.API, error) {
 
 	cf, err := broker.GetClient()

broker/create_registry_server_instance.go

@@ -1,12 +1,10 @@
 package broker
 
 import (
-	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
 	"path"
-	"strings"
 
 	"code.cloudfoundry.org/cli/api/cloudcontroller/ccv3"
 	"code.cloudfoundry.org/cli/api/cloudcontroller/ccv3/constant"
@@ -19,6 +17,18 @@ func (broker *SCSBroker) createRegistryServerInstance(serviceId string, instance
 		return "", err
 	}
 
+	//rc := utilities.NewRegistryConfig()
+	broker.Logger.Info("jsonparams == " + jsonparams)
+	//rp, err := utilities.ExtractRegistryParams(jsonparams)
+	//if err != nil {
+	//return "", err
+	//}
+
+	//count, err := rp.Count()
+	//if err != nil {
+	//return "", err
+	//}
+
 	cfClient, err := broker.GetClient()
 	if err != nil {
 		return "", errors.New("Couldn't start session: " + err.Error())
@@ -134,38 +144,39 @@ func (broker *SCSBroker) createRegistryServerInstance(serviceId string, instance
 		return "", err
 	}
 
-	broker.Logger.Info("handle node count")
-	// handle the node count
-	rc := &registryConfig{}
-	broker.Logger.Info("jsonparams == " + jsonparams)
-	rp, err := utilities.ExtractRegistryParams(jsonparams)
-	if err != nil {
-		return "", err
-	}
+	//broker.Logger.Info("handle node count")
+	//// handle the node count
+	//if count > 1 {
+	//rc.Clustered()
+	//broker.Logger.Info(fmt.Sprintf("scaling to %d", count))
+	//err = broker.scaleRegistryServer(cfClient, &app, count)
+	//if err != nil {
+	//return "", err
+	//}
 
-	if count, found := rp["count"]; found {
-		if c, ok := count.(int); ok {
-			if c > 1 {
-				rc.Clustered()
-				broker.Logger.Info(fmt.Sprintf("scaling to %d", c))
-				err = broker.scaleRegistryServer(cfClient, &app, c, rc)
-				if err != nil {
-					return "", err
-				}
-			} else {
-				rc.Standalone()
-			}
-		} else {
-			rc.Standalone()
-		}
-	}
+	//community, err := broker.GetCommunity()
+	//if err != nil {
+	//return "", err
+	//}
 
-	broker.Logger.Info("Updating Environment")
-	err = broker.UpdateAppEnvironment(cfClient, &app, &info, serviceId, instanceId, rc.String(), params)
+	//stats, err := getProcessStatsByAppAndType(cfClient, community, broker.Logger, app.GUID, "web")
+	//if err != nil {
+	//return "", nil
+	//}
 
-	if err != nil {
-		return "", err
-	}
+	//for _, stat := range stats {
+	//rc.AddPeer(stat.Index, "http", stat.Host, stat.InstancePorts[0].External)
+	//}
+	//} else {
+	//rc.Standalone()
+	//}
+
+	//broker.Logger.Info("Updating Environment")
+	//err = broker.UpdateRegistryEnvironment(cfClient, &app, &info, serviceId, instanceId, rc, params)
+
+	//if err != nil {
+	//return "", err
+	//}
 
 	app, _, err = cfClient.UpdateApplicationRestart(app.GUID)
 	if err != nil {
@@ -176,54 +187,3 @@ func (broker *SCSBroker) createRegistryServerInstance(serviceId string, instance
 
 	return route.URL, nil
 }
-
-type registryConfig struct {
-	Mode  string
-	Peers []string
-}
-
-func (rc *registryConfig) AddPeer(peer string) {
-	rc.Peers = append(rc.Peers, peer)
-}
-
-func (rc *registryConfig) Standalone() {
-	rc.Mode = "standalone"
-}
-
-func (rc *registryConfig) Clustered() {
-	rc.Mode = "clustered"
-}
-
-func (rc *registryConfig) String() string {
-	return string(rc.Bytes())
-}
-
-func (rc *registryConfig) Bytes() []byte {
-	client := make(map[string]interface{})
-
-	if rc.Mode == "standalone" {
-		client["registerWithEureka"] = false
-		client["fetchRegistry"] = false
-	}
-
-	if len(rc.Peers) > 0 {
-		serviceUrl := make(map[string]interface{})
-		defaultZone := strings.Join(rc.Peers, ",")
-		serviceUrl["defaultZone"] = defaultZone
-		client["serviceUrl"] = serviceUrl
-	}
-
-	eureka := make(map[string]interface{})
-	eureka["client"] = client
-
-	data := make(map[string]interface{})
-	data["eureka"] = eureka
-
-	output, err := json.Marshal(data)
-	if err != nil {
-		return []byte("{}")
-	}
-
-	return output
-
-}

broker/deprovision.go

@@ -4,7 +4,7 @@ import (
 	"context"
 
 	"code.cloudfoundry.org/cli/api/cloudcontroller/ccerror"
-	brokerapi "github.com/pivotal-cf/brokerapi/domain"
+	brokerapi "github.com/pivotal-cf/brokerapi/v7/domain"
 	"github.com/starkandwayne/scs-broker/broker/utilities"
 )
 

broker/get_process_stats_by_app.go

@@ -0,0 +1,39 @@
+package broker
+
+import (
+	"errors"
+
+	cf "github.com/cloudfoundry-community/go-cfclient"
+)
+
+func (broker *SCSBroker) getProcessStatsByAppAndType(appGUID string, procType string) ([]cf.Stats, error) {
+	stats := make([]cf.Stats, 0)
+
+	community, err := broker.GetCommunity()
+	if err != nil {
+		return stats, err
+	}
+
+	procs, err := broker.getApplicationProcessesByType(appGUID, procType)
+	if err != nil {
+		return stats, err
+	}
+
+	for _, proc := range procs {
+		candidates, err := community.GetProcessStats(proc.GUID)
+		if err != nil {
+			continue
+		}
+
+		for _, stat := range candidates {
+			stats = append(stats, stat)
+		}
+	}
+
+	if len(stats) == 0 {
+		return stats, errors.New("no stats found")
+	}
+
+	return stats, nil
+
+}

broker/log_workflow_error.go

@@ -0,0 +1,7 @@
+package broker
+
+import "code.cloudfoundry.org/lager"
+
+func (broker *SCSBroker) logWorkflowError(msg string, workflow string, err error) {
+	broker.Logger.Info(msg, lager.Data{"workflow": workflow, "error": err.Error()})
+}

broker/poll_apply_manifest.go

@@ -0,0 +1,75 @@
+package broker
+
+import (
+	"time"
+
+	"code.cloudfoundry.org/cli/api/cloudcontroller/ccv3"
+)
+
+//func (broker *SCSBroker) pollApplyManifest(jobURL ccv3.JobURL) (ccv3.Warnings, error) {
+//var allWarnings ccv3.Warnings
+//cfClient, err := broker.GetClient()
+//if err != nil {
+//return nil, errors.New("Couldn't start session: " + err.Error())
+//}
+//count := 0
+
+//for {
+//time.Sleep(time.Second)
+//count += 1
+
+//broker.Logger.Info(fmt.Sprintf("Polling iteration %d, job %s", count, jobURL))
+//job, warnings, err := cfClient.GetJob(jobURL)
+//allWarnings = append(allWarnings, warnings...)
+//broker.showWarnings(warnings, "poll-apply-manifest")
+//if err != nil {
+//return allWarnings, err
+//}
+
+//broker.Logger.Info(fmt.Sprintf("HERE'S THE FUCKING JOB STATE: %s", job.State))
+//if job.HasFailed() {
+//err = job.Errors()[0]
+//broker.logWorkflowError("pollApplyManifest", "*none*", err)
+//return allWarnings, err
+//}
+
+//if job.IsComplete() {
+//return allWarnings, nil
+//}
+//}
+
+//return allWarnings, nil
+//}
+
+func (broker *SCSBroker) pollApplyManifest(jobURL ccv3.JobURL) (ccv3.Warnings, error) {
+	var (
+		err         error
+		warnings    ccv3.Warnings
+		allWarnings ccv3.Warnings
+		job         ccv3.Job
+	)
+
+	client, err := broker.GetClient()
+	if err != nil {
+		return allWarnings, err
+	}
+
+	for {
+		job, warnings, err = client.GetJob(jobURL)
+		allWarnings = append(allWarnings, warnings...)
+		if err != nil {
+			return allWarnings, err
+		}
+
+		if job.HasFailed() {
+			firstError := job.Errors()[0]
+			return allWarnings, firstError
+		}
+
+		if job.IsComplete() {
+			return allWarnings, nil
+		}
+
+		time.Sleep(5 * time.Second)
+	}
+}