Opt

pkg/controller.go

@@ -62,7 +62,7 @@ type Controller struct {
 func NewController(consulAddress string) *Controller {
 	controller := &Controller{
 		consulAddress: consulAddress,
-		pushChannel:   make(chan *ChangeEvent),
+		pushChannel:   make(chan *ChangeEvent, 1),
 	}
 	return controller
 }
@@ -129,7 +129,7 @@ func (s *Controller) mainLoop(stop <-chan struct{}) {
 					if err != nil {
 						log.Errorf("Failed to synchronize consul services to Istio: %v", err)
 						// Retry if failed
-						s.pushChannel <- &ChangeEvent{}
+						// s.pushChannel <- &ChangeEvent{}
 					}
 					debouncedEvents = 0
 				}

pkg/serviceregistry/consul/controller.go

@@ -85,7 +85,7 @@ func (c *Controller) initCache() error {
 	if err != nil {
 		return err
 	}
-
+	c.servicesList = nil
 	for serviceName := range consulServices {
 		// get endpoints of a service from consul
 		endpoints, err := c.getCatalogService(serviceName, nil)

pkg/serviceregistry/consul/conversion.go

@@ -42,7 +42,8 @@ func convertServiceEntry(service string, endpoints []*api.CatalogService) *istio
 	for _, endpoint := range endpoints {
 		name = endpoint.ServiceName
 
-		port := convertPort(endpoint.ServicePort, endpoint.ServiceMeta[protocolTagName])
+		protoName := getProtoValFromServiceTagsOrMeta(endpoint.ServiceTags, endpoint.ServiceMeta[protocolTagName])
+		port := convertPort(endpoint.ServicePort, protoName)
 
 		if svcPort, exists := ports[port.Number]; exists && svcPort.Protocol != port.Protocol {
 			log.Warnf("Service %v has two instances on same port %v but different protocols (%v, %v)",
@@ -83,7 +84,9 @@ func convertWorkloadEntry(endpoint *api.CatalogService) *istio.WorkloadEntry {
 	if addr == "" {
 		addr = endpoint.Address
 	}
-	port := convertPort(endpoint.ServicePort, endpoint.ServiceMeta[protocolTagName])
+
+	protoName := getProtoValFromServiceTagsOrMeta(endpoint.ServiceTags, endpoint.ServiceMeta[protocolTagName])
+	port := convertPort(endpoint.ServicePort, protoName)
 
 	return &istio.WorkloadEntry{
 		Address:  addr,
@@ -135,3 +138,27 @@ func convertProtocol(name string) string {
 	}
 	return string(p)
 }
+
+//get the protocol value:get from meta first, otherwise get from tags
+func getProtoValFromServiceTagsOrMeta(serviceTags []string, serviceMetaProtocol string) string {
+	if len(serviceMetaProtocol) > 0 {
+		log.Debugf("getProtoValFromServiceTagsOrMeta,serviceMetaProtocol:%v", serviceMetaProtocol)
+		return serviceMetaProtocol
+	}
+	defaultProto := "http"
+	if len(serviceTags) == 0 {
+		return defaultProto
+	}
+	for _, v := range serviceTags {
+		isExist := strings.Contains(v, "proto")
+		if isExist {
+			strs := strings.Split(v, "=")
+			if len(strs) == 2 {
+				return strs[1]
+			} else {
+				log.Warnf("service tags proto is invalid")
+			}
+		}
+	}
+	return defaultProto
+}

pkg/serviceregistry/consul/monitor.go

@@ -37,7 +37,10 @@ type consulMonitor struct {
 	ServiceChangeHandlers []ServiceChangeHandler
 }
 
-const blockQueryWaitTime time.Duration = 10 * time.Minute
+const (
+	blockQueryWaitTime          time.Duration = 10 * time.Minute
+	updateServiceRecordWaitTime time.Duration = 10 * time.Second
+)
 
 // NewConsulMonitor watches for changes in Consul services and CatalogServices
 func NewConsulMonitor(client *api.Client) Monitor {
@@ -72,6 +75,7 @@ func (m *consulMonitor) watchConsul(stop <-chan struct{}) {
 			} else if consulWaitIndex != queryMeta.LastIndex {
 				consulWaitIndex = queryMeta.LastIndex
 				m.updateServiceRecord()
+				time.Sleep(updateServiceRecordWaitTime)
 			}
 		}
 	}