Add emctl handling consul service registry and cookbook of it

docs/cookbook/README.md

@@ -0,0 +1,3 @@
+# Cookbook for Common User Cases
+
+- [External Service Registry](./external_service_registry.md) - Co-exist with external service registry.

docs/cookbook/external_service_registry.md

@@ -0,0 +1,53 @@
+
+# External Service Registry
+
+When architecture style moves to service mesh, there could be a middle status for the technical stack. Which is that there would be both legacy services and mesh services co-existing. The two different kinds of services also want to call each other.
+
+EaseMesh uses internal component Etcd in the control plane to play the role of service registry, and the external services may use Consul, Nacos, Eureka, Zookeeper as a service registry. Based on service registry controllers from Easegress, we develop a registry syncer to synchronize services between internal Etcd and the external service registry.
+
+As an example, we use `emctl` to apply consul service registry:
+
+```bash
+$ emctl apply -f consul-service-registry.yaml
+```
+
+consul-service-registry.yaml
+
+```yaml
+apiVersion: mesh.megaease.com/v1alpha1
+kind: ConsulServiceRegistry
+metadata:
+  name: consul-service-registry
+address: consul-server-0.consul-server.default:8500
+scheme: http
+datacenter: ""
+token: ""
+namespace: ""
+syncInterval: 5s
+serviceTags: []
+```
+
+Then we need to specify external service registry name in MeshController:
+
+```bash
+$ emctl apply -f mesh-controller.yaml
+```
+
+mesh-controller.yaml
+
+```yaml
+apiVersion: mesh.megaease.com/v1alpha1
+kind: MeshController
+metadata:
+  name: easemesh-controller
+apiPort: 13009
+ingressPort: 19527
+heartbeatInterval: 5s
+# +
+externalServiceRegistry: consul-service-registry
+registryType: consul
+```
+
+And we could use `emctl get service && emctl get serviceinstance` to query the service and instance information from different sources.
+
+> NOTICE: The registry syncer transforms the service entry format between different registries. The connectability needs to be guaranteed by themselves. For example, services run in the same Kubernetes cluster would be connectable without other operations.

emctl/cmd/client/command/apply/applier.go

@@ -46,6 +46,8 @@ func WrapApplierByMeshObject(object meta.MeshObject,
 	switch object.Kind() {
 	case resource.KindMeshController:
 		return &meshControllerApplier{object: object.(*resource.MeshController), baseApplier: baseApplier{client: client, timeout: timeout}}
+	case resource.KindConsulServiceRegistry:
+		return &consulServiceRegistryApplier{object: object.(*resource.ConsulServiceRegistry), baseApplier: baseApplier{client: client, timeout: timeout}}
 	case resource.KindService:
 		return &serviceApplier{object: object.(*resource.Service), baseApplier: baseApplier{client: client, timeout: timeout}}
 	case resource.KindServiceInstance:
@@ -103,6 +105,36 @@ func (mc *meshControllerApplier) Apply() error {
 	}
 }
 
+type consulServiceRegistryApplier struct {
+	baseApplier
+	object *resource.ConsulServiceRegistry
+}
+
+func (c *consulServiceRegistryApplier) Apply() error {
+	ctx, cancelFunc := context.WithTimeout(context.Background(), c.timeout)
+	defer cancelFunc()
+	err := c.client.V1Alpha1().ConsulServiceRegistry().Create(ctx, c.object)
+	for {
+		switch {
+		case err == nil:
+			return nil
+		case meshclient.IsConflictError(err):
+			err = c.client.V1Alpha1().ConsulServiceRegistry().Patch(ctx, c.object)
+			if err != nil {
+				return errors.Wrapf(err, "update consulServiceRegistry %s", c.object.Name())
+			}
+		case meshclient.IsNotFoundError(err):
+			err = c.client.V1Alpha1().ConsulServiceRegistry().Create(ctx, c.object)
+			if err != nil {
+				return errors.Wrapf(err, "create consulServiceRegistry %s", c.object.Name())
+			}
+		default:
+			return errors.Wrapf(err, "apply consulServiceRegistry %s", c.object.Name())
+		}
+
+	}
+}
+
 type serviceApplier struct {
 	baseApplier
 	object *resource.Service

emctl/cmd/client/command/delete/deleter.go

@@ -34,6 +34,8 @@ func WrapDeleterByMeshObject(object meta.MeshObject,
 	switch object.Kind() {
 	case resource.KindMeshController:
 		return &meshControllerDeleter{object: object.(*resource.MeshController), baseDeleter: baseDeleter{client: client, timeout: timeout}}
+	case resource.KindConsulServiceRegistry:
+		return &consulServiceRegistryDeleter{object: object.(*resource.ConsulServiceRegistry), baseDeleter: baseDeleter{client: client, timeout: timeout}}
 	case resource.KindService:
 		return &serviceDeleter{object: object.(*resource.Service), baseDeleter: baseDeleter{client: client, timeout: timeout}}
 	case resource.KindServiceInstance:
@@ -82,6 +84,17 @@ func (s *meshControllerDeleter) Delete() error {
 	return s.client.V1Alpha1().MeshController().Delete(ctx, s.object.Name())
 }
 
+type consulServiceRegistryDeleter struct {
+	baseDeleter
+	object *resource.ConsulServiceRegistry
+}
+
+func (c *consulServiceRegistryDeleter) Delete() error {
+	ctx, cancelFunc := context.WithTimeout(context.Background(), c.timeout)
+	defer cancelFunc()
+	return c.client.V1Alpha1().ConsulServiceRegistry().Delete(ctx, c.object.Name())
+}
+
 type serviceDeleter struct {
 	baseDeleter
 	object *resource.Service

emctl/cmd/client/command/get/getter.go

@@ -38,6 +38,8 @@ func WrapGetterByMeshObject(object meta.MeshObject,
 	switch object.Kind() {
 	case resource.KindMeshController:
 		return &meshControllerGetter{object: object.(*resource.MeshController), baseGetter: base}
+	case resource.KindConsulServiceRegistry:
+		return &consulServiceRegistryGetter{object: object.(*resource.ConsulServiceRegistry), baseGetter: base}
 	case resource.KindService:
 		return &serviceGetter{object: object.(*resource.Service), baseGetter: base}
 	case resource.KindServiceInstance:
@@ -108,6 +110,37 @@ func (s *meshControllerGetter) Get() ([]meta.MeshObject, error) {
 	return objects, nil
 }
 
+type consulServiceRegistryGetter struct {
+	baseGetter
+	object *resource.ConsulServiceRegistry
+}
+
+func (s *consulServiceRegistryGetter) Get() ([]meta.MeshObject, error) {
+	ctx, cancelFunc := context.WithTimeout(context.Background(), s.timeout)
+	defer cancelFunc()
+
+	if s.object.Name() != "" {
+		consulServiceRegistry, err := s.client.V1Alpha1().ConsulServiceRegistry().Get(ctx, s.object.Name())
+		if err != nil {
+			return nil, err
+		}
+
+		return []meta.MeshObject{consulServiceRegistry}, nil
+	}
+
+	consulServiceRegistrys, err := s.client.V1Alpha1().ConsulServiceRegistry().List(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	objects := make([]meta.MeshObject, len(consulServiceRegistrys))
+	for i := range consulServiceRegistrys {
+		objects[i] = consulServiceRegistrys[i]
+	}
+
+	return objects, nil
+}
+
 type serviceGetter struct {
 	baseGetter
 	object *resource.Service

emctl/cmd/client/command/meshclient/consts.go

@@ -26,6 +26,12 @@ const (
 	// MeshControllerURL is the mesh controller path.
 	MeshControllerURL = apiURL + "/objects/%s"
 
+	// ConsulServiceRegistrysURL is the consul service registry path.
+	ConsulServiceRegistrysURL = apiURL + "/objects"
+
+	// ConsulServiceRegistryURL is the consul service registry path.
+	ConsulServiceRegistryURL = apiURL + "/objects/%s"
+
 	// MeshTenantsURL is the mesh tenant prefix.
 	MeshTenantsURL = apiURL + "/mesh/tenants"
 

emctl/cmd/client/command/meshclient/consulserviceregistry.go

@@ -0,0 +1,176 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * 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 meshclient
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/megaease/easemeshctl/cmd/client/resource"
+	"github.com/megaease/easemeshctl/cmd/common/client"
+	"gopkg.in/yaml.v2"
+
+	"github.com/pkg/errors"
+)
+
+type consulServiceRegistryGetter struct {
+	client *meshClient
+}
+
+func (t *consulServiceRegistryGetter) ConsulServiceRegistry() ConsulServiceRegistryInterface {
+	return &consulServiceRegistryInterface{client: t.client}
+}
+
+type consulServiceRegistryInterface struct {
+	client *meshClient
+}
+
+func (t *consulServiceRegistryInterface) Get(ctx context.Context, consulServiceRegistryID string) (*resource.ConsulServiceRegistry, error) {
+	url := fmt.Sprintf("http://"+t.client.server+ConsulServiceRegistryURL, consulServiceRegistryID)
+	re, err := client.NewHTTPJSON().
+		GetByContext(ctx, url, nil, nil).
+		HandleResponse(func(b []byte, statusCode int) (interface{}, error) {
+			if statusCode == http.StatusNotFound {
+				return nil, errors.Wrapf(NotFoundError, "get consulServiceRegistry %s", consulServiceRegistryID)
+			}
+
+			if statusCode >= 300 {
+				return nil, errors.Errorf("call %s failed: return status code: %d text: %s", url, statusCode, string(b))
+			}
+			consulServiceRegistry := &resource.ConsulServiceRegistryV1Alpha1{}
+			err := yaml.Unmarshal(b, consulServiceRegistry)
+			if err != nil {
+				return nil, errors.Wrap(err, "unmarshal data to ConsulServiceRegistry")
+			}
+			return resource.ToConsulServiceRegistry(consulServiceRegistry), nil
+		})
+	if err != nil {
+		return nil, err
+	}
+
+	return re.(*resource.ConsulServiceRegistry), nil
+}
+
+func (t *consulServiceRegistryInterface) Patch(ctx context.Context, consulServiceRegistry *resource.ConsulServiceRegistry) error {
+	url := fmt.Sprintf("http://"+t.client.server+ConsulServiceRegistryURL, consulServiceRegistry.Name())
+	update, err := yaml.Marshal(consulServiceRegistry.ToV1Alpha1())
+	if err != nil {
+		return fmt.Errorf("marshal %#v to yaml failed: %v", consulServiceRegistry, err)
+	}
+
+	_, err = client.NewHTTPJSON().
+		PutByContext(ctx, url, update, nil).
+		HandleResponse(func(b []byte, statusCode int) (interface{}, error) {
+			if statusCode == http.StatusNotFound {
+				return nil, errors.Wrapf(NotFoundError, "patch consulServiceRegistry %s", consulServiceRegistry.Name())
+			}
+
+			if statusCode < 300 && statusCode >= 200 {
+				return nil, nil
+			}
+			return nil, errors.Errorf("call PUT %s failed, return statuscode %d text %s", url, statusCode, string(b))
+		})
+
+	return err
+}
+
+func (t *consulServiceRegistryInterface) Create(ctx context.Context, consulServiceRegistry *resource.ConsulServiceRegistry) error {
+	url := fmt.Sprintf("http://" + t.client.server + ConsulServiceRegistrysURL)
+	create, err := yaml.Marshal(consulServiceRegistry.ToV1Alpha1())
+	if err != nil {
+		return fmt.Errorf("marshal %#v to yaml failed: %v", consulServiceRegistry, err)
+	}
+
+	_, err = client.NewHTTPJSON().
+		PostByContext(ctx, url, create, nil).
+		HandleResponse(func(b []byte, statusCode int) (interface{}, error) {
+			if statusCode == http.StatusConflict {
+				return nil, errors.Wrapf(ConflictError, "create consulServiceRegistry %s", consulServiceRegistry.Name())
+			}
+
+			if statusCode < 300 && statusCode >= 200 {
+				return nil, nil
+			}
+			return nil, errors.Errorf("call Post %s failed, return statuscode %d text %s", url, statusCode, string(b))
+		})
+
+	return err
+}
+
+func (t *consulServiceRegistryInterface) Delete(ctx context.Context, consulServiceRegistryID string) error {
+	url := fmt.Sprintf("http://"+t.client.server+ConsulServiceRegistryURL, consulServiceRegistryID)
+	_, err := client.NewHTTPJSON().
+		DeleteByContext(ctx, url, nil, nil).
+		HandleResponse(func(b []byte, statusCode int) (interface{}, error) {
+			if statusCode == http.StatusNotFound {
+				return nil, NotFoundError
+			}
+			if statusCode < 300 && statusCode >= 200 {
+				return nil, nil
+			}
+			return nil, errors.Errorf("call DELETE %s failed, return statuscode %d text %s", url, statusCode, string(b))
+		})
+	return err
+}
+
+func (t *consulServiceRegistryInterface) List(ctx context.Context) ([]*resource.ConsulServiceRegistry, error) {
+	url := fmt.Sprintf("http://" + t.client.server + ConsulServiceRegistrysURL)
+	result, err := client.NewHTTPJSON().
+		GetByContext(ctx, url, nil, nil).
+		HandleResponse(func(b []byte, statusCode int) (interface{}, error) {
+			if statusCode == http.StatusNotFound {
+				return nil, errors.Wrap(NotFoundError, "list consulServiceRegistry")
+			}
+
+			if statusCode >= 300 || statusCode < 200 {
+				return nil, errors.Errorf("call GET %s failed, return statuscode %d text %s", url, statusCode, string(b))
+			}
+
+			objects := []map[string]interface{}{}
+			err := yaml.Unmarshal(b, &objects)
+			if err != nil {
+				return nil, errors.Wrapf(err, "unmarshal objects")
+			}
+
+			results := []*resource.ConsulServiceRegistry{}
+			for _, object := range objects {
+				if object["kind"] != resource.KindConsulServiceRegistry {
+					continue
+				}
+
+				buff, err := yaml.Marshal(object)
+				if err != nil {
+					return nil, errors.Wrapf(err, "marshal %#v to yaml", object)
+				}
+
+				consulServiceRegistry := &resource.ConsulServiceRegistryV1Alpha1{}
+				err = yaml.Unmarshal(buff, consulServiceRegistry)
+				if err != nil {
+					return nil, fmt.Errorf("unmarshal %s to yaml failed: %v", buff, err)
+				}
+
+				results = append(results, resource.ToConsulServiceRegistry(consulServiceRegistry))
+			}
+			return results, nil
+		})
+	if err != nil {
+		return nil, err
+	}
+	return result.([]*resource.ConsulServiceRegistry), err
+}