foundations/foundation-registry/src/main/java/org/apache/servicecomb/registry/consumer/MicroserviceVersions.java - servicecomb-java-chassis - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You 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 org.apache.servicecomb.registry.consumer;

 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.stream.Collectors;

 import org.apache.servicecomb.foundation.common.VendorExtensions;
 import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
 import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
 import org.apache.servicecomb.registry.DiscoveryManager;
 import org.apache.servicecomb.registry.api.event.CreateMicroserviceEvent;
 import org.apache.servicecomb.registry.api.event.DestroyMicroserviceEvent;
 import org.apache.servicecomb.registry.api.event.MicroserviceInstanceChangedEvent;
 import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
 import org.apache.servicecomb.registry.api.registry.MicroserviceInstanceStatus;
 import org.apache.servicecomb.registry.api.registry.MicroserviceInstances;
 import org.apache.servicecomb.registry.config.ServiceRegistryCommonConfig;
 import org.apache.servicecomb.registry.definition.DefinitionConst;
 import org.apache.servicecomb.registry.definition.MicroserviceNameParser;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 public class MicroserviceVersions {
   private static final Logger LOGGER = LoggerFactory.getLogger(MicroserviceVersions.class);

   protected AppManager appManager;

   protected String appId;

   protected String shortName;

   protected String microserviceName;

   // revision and pulledInstances directly equals to SC's response
   protected String revision = null;

   private List<MicroserviceInstance> pulledInstances;

   private MicroserviceInstances lastPulledResult;

   // instances not always equals to pulledInstances
   // in the future:
   //  pulledInstances means all instance
   //  instances means available instance
   Collection<MicroserviceInstance> instances;

   // key is service id
   Map<String, MicroserviceVersion> versions = new ConcurrentHashMapEx<>();

   // key is version rule
   Map<String, MicroserviceVersionRule> versionRules = new ConcurrentHashMapEx<>();

   // process pulled instances and create versionRule must be protected by lock
   // otherwise maybe lost instance or version in versionRule
   private final Object lock = new Object();

   private long lastPullTime = 0;

   private boolean waitingDelete = false;

   private final VendorExtensions vendorExtensions = new VendorExtensions();

   public MicroserviceVersions(AppManager appManager, String appId, String microserviceName) {
     this.appManager = appManager;
     this.appId = appId;
     this.microserviceName = microserviceName;
     this.shortName = new MicroserviceNameParser(appId, microserviceName).getShortName();

     appManager.getEventBus().post(new CreateMicroserviceEvent(this));

     LOGGER.info("create MicroserviceVersions, appId={}, microserviceName={}.",
         appId,
         microserviceName);
   }

   public boolean isWaitingDelete() {
     return waitingDelete;
   }

   public MicroserviceVersions markWaitingDelete() {
     this.waitingDelete = true;
     return this;
   }

   public AppManager getAppManager() {
     return appManager;
   }

   public String getAppId() {
     return appId;
   }

   public String getMicroserviceName() {
     return microserviceName;
   }

   public String getShortName() {
     return shortName;
   }

   public Map<String, MicroserviceVersion> getVersions() {
     return versions;
   }

   @SuppressWarnings("unchecked")
   public <T extends MicroserviceVersion> T getVersion(String serviceId) {
     return (T) versions.get(serviceId);
   }

   public String getRevision() {
     return revision;
   }

   public void setRevision(String revision) {
     this.revision = revision;
   }

   public List<MicroserviceInstance> getPulledInstances() {
     return pulledInstances;
   }

   public long getLastPullTime() {
     return lastPullTime;
   }

   public MicroserviceInstances getLastPulledResult() {
     return lastPulledResult;
   }

   public VendorExtensions getVendorExtensions() {
     return vendorExtensions;
   }

   public void pullInstances() {
     lastPullTime = System.currentTimeMillis();
     MicroserviceInstances microserviceInstances = findServiceInstances();
     lastPulledResult = microserviceInstances;
     if (microserviceInstances == null) {
       // pulled failed, did not get anything
       // will not do anything, consumers will use existing instances
       return;
     }
     if (microserviceInstances.isMicroserviceNotExist()) {
       // pulled failed, SC said target not exist
       waitingDelete = true;
       return;
     }

     if (null != revision && revision.equals(microserviceInstances.getRevision())) {
       return;
     }

     pulledInstances = microserviceInstances.getInstancesResponse().getInstances();
     pulledInstances.sort(Comparator.comparing(MicroserviceInstance::getInstanceId));
     String rev = microserviceInstances.getRevision();

     safeSetInstances(pulledInstances, rev);
   }

   protected MicroserviceInstances findServiceInstances() {
     return DiscoveryManager.INSTANCE.findServiceInstances(appId,
         microserviceName,
         DefinitionConst.VERSION_RULE_ALL,
         revision);
   }

   protected void safeSetInstances(List<MicroserviceInstance> pulledInstances, String rev) {
     try {
       List<MicroserviceInstance> filteredInstance = pulledInstances;
       // 增加一个配置项只使用 `UP` 实例。 在使用 `TESTING` 进行拨测， 并且配置了
       // servicecomb.references.version-rule=latest 场景，需要保证不使用
       // `TESTING` 实例。 不能依赖 InstanceStatusDiscoveryFilter, 避免
       // 构建的 VersionRule 实例列表为空。
       if (ServiceRegistryCommonConfig.useUpInstancesOnly()) {
         filteredInstance = pulledInstances.stream().filter(item -> MicroserviceInstanceStatus.UP == item.getStatus())
             .collect(Collectors.toList());
       }
       setInstances(filteredInstance, rev);
     } catch (Throwable e) {
       waitingDelete = true;
       LOGGER.error("Failed to setInstances, appId={}, microserviceName={}.",
           getAppId(),
           getMicroserviceName(),
           e);
     }
   }

   static class MergedInstances {
     // key is microserviceId
     Map<String, List<MicroserviceInstance>> microserviceIdMap = new HashMap<>();

     // key is instanceId
     Map<String, MicroserviceInstance> instanceIdMap = new HashMap<>();

     void addInstance(MicroserviceInstance instance) {
       instanceIdMap.put(instance.getInstanceId(), instance);
       microserviceIdMap
           .computeIfAbsent(instance.getServiceId(), key -> new ArrayList<>())
           .add(instance);
     }
   }

   private void setInstances(List<MicroserviceInstance> pulledInstances, String rev) {
     synchronized (lock) {
       MergedInstances mergedInstances = mergeInstances(pulledInstances, instances);
       instances = mergedInstances.instanceIdMap.values();
       // clear cache
       versions.forEach((key, value) -> value.setInstances(new ArrayList<>()));
       for (Entry<String, List<MicroserviceInstance>> entry : mergedInstances.microserviceIdMap.entrySet()) {
         // ensure microserviceVersion exists
         versions.computeIfAbsent(entry.getKey(),
             microserviceId -> createMicroserviceVersion(microserviceId, entry.getValue()))
             .setInstances(entry.getValue());
       }

       for (MicroserviceVersionRule microserviceVersionRule : versionRules.values()) {
         microserviceVersionRule.update(versions, instances);
       }
       revision = rev;
     }
   }

   protected MicroserviceVersion createMicroserviceVersion(String microserviceId, List<MicroserviceInstance> instances) {
     return new MicroserviceVersion(this, microserviceId, microserviceName, instances);
   }

   private MergedInstances mergeInstances(List<MicroserviceInstance> pulledInstances,
       Collection<MicroserviceInstance> inUseInstances) {
     MergedInstances mergedInstances = new MergedInstances();
     pulledInstances.stream().forEach(mergedInstances::addInstance);
     MicroserviceInstancePing ping = SPIServiceUtils.getPriorityHighestService(MicroserviceInstancePing.class);

     if (pulledInstances.isEmpty() && inUseInstances != null && ServiceRegistryCommonConfig
         .isEmptyInstanceProtectionEnabled()) {

       inUseInstances.stream().forEach(instance -> {
         if (!mergedInstances.instanceIdMap.containsKey(instance.getInstanceId())) {
           if (ping.ping(instance)) {
             mergedInstances.addInstance(instance);
           }
         }
       });
     }

     return mergedInstances;
   }

   public MicroserviceVersionRule getOrCreateMicroserviceVersionRule(String versionRule) {
     // do not use computeIfAbsent
     MicroserviceVersionRule microserviceVersionRule = versionRules.get(versionRule);
     if (microserviceVersionRule == null) {
       synchronized (lock) {
         microserviceVersionRule = versionRules.computeIfAbsent(versionRule, this::createAndInitMicroserviceVersionRule);
       }
     }

     return microserviceVersionRule;
   }

   protected MicroserviceVersionRule createAndInitMicroserviceVersionRule(String strVersionRule) {
     LOGGER.info("create MicroserviceVersionRule, appId={}, microserviceName={}, versionRule={}.",
         appId,
         microserviceName,
         strVersionRule);

     MicroserviceVersionRule microserviceVersionRule =
         new MicroserviceVersionRule(appId, microserviceName, strVersionRule);
     microserviceVersionRule.update(versions, instances);
     return microserviceVersionRule;
   }

   public void onMicroserviceInstanceChanged(MicroserviceInstanceChangedEvent changedEvent) {
     if (!isEventAccept(changedEvent)) {
       return;
     }
     // pull instances always replace old instances, not append
     //
     // pull result and watch event sequence is not defined even inside SC.
     // it's not safe to trust the event, so we just send a new pull request
     //
     // CREATE/UPDATE:
     //   if pull 1/2/3, and then add 4, but "add 4" received before pull result, will lost 4.
     // DELETE:
     //   if pull 1/2/3, and then delete 3, but "delete 3" received before pull result, will have wrong 3.
     // EXPIRE::
     //   black/white config in SC changed, we must refresh all data from sc.
     pullInstances();
   }

   protected boolean isEventAccept(MicroserviceInstanceChangedEvent changedEvent) {
     return (appId.equals(changedEvent.getKey().getAppId()) &&
         microserviceName.equals(changedEvent.getKey().getServiceName())) ||
         microserviceName.equals(
             changedEvent.getKey().getAppId() + DefinitionConst.APP_SERVICE_SEPARATOR + changedEvent.getKey()
                 .getServiceName());
   }

   public void destroy() {
     for (MicroserviceVersion microserviceVersion : versions.values()) {
       microserviceVersion.destroy();
     }

     appManager.getEventBus().post(new DestroyMicroserviceEvent(this));
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You 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 org.apache.servicecomb.registry.consumer;

	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Comparator;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Map.Entry;
	import java.util.stream.Collectors;

	import org.apache.servicecomb.foundation.common.VendorExtensions;
	import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
	import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
	import org.apache.servicecomb.registry.DiscoveryManager;
	import org.apache.servicecomb.registry.api.event.CreateMicroserviceEvent;
	import org.apache.servicecomb.registry.api.event.DestroyMicroserviceEvent;
	import org.apache.servicecomb.registry.api.event.MicroserviceInstanceChangedEvent;
	import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
	import org.apache.servicecomb.registry.api.registry.MicroserviceInstanceStatus;
	import org.apache.servicecomb.registry.api.registry.MicroserviceInstances;
	import org.apache.servicecomb.registry.config.ServiceRegistryCommonConfig;
	import org.apache.servicecomb.registry.definition.DefinitionConst;
	import org.apache.servicecomb.registry.definition.MicroserviceNameParser;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	public class MicroserviceVersions {
	private static final Logger LOGGER = LoggerFactory.getLogger(MicroserviceVersions.class);

	protected AppManager appManager;

	protected String appId;

	protected String shortName;

	protected String microserviceName;

	// revision and pulledInstances directly equals to SC's response
	protected String revision = null;

	private List<MicroserviceInstance> pulledInstances;

	private MicroserviceInstances lastPulledResult;

	// instances not always equals to pulledInstances
	// in the future:
	// pulledInstances means all instance
	// instances means available instance
	Collection<MicroserviceInstance> instances;

	// key is service id
	Map<String, MicroserviceVersion> versions = new ConcurrentHashMapEx<>();

	// key is version rule
	Map<String, MicroserviceVersionRule> versionRules = new ConcurrentHashMapEx<>();

	// process pulled instances and create versionRule must be protected by lock
	// otherwise maybe lost instance or version in versionRule
	private final Object lock = new Object();

	private long lastPullTime = 0;

	private boolean waitingDelete = false;

	private final VendorExtensions vendorExtensions = new VendorExtensions();

	public MicroserviceVersions(AppManager appManager, String appId, String microserviceName) {
	this.appManager = appManager;
	this.appId = appId;
	this.microserviceName = microserviceName;
	this.shortName = new MicroserviceNameParser(appId, microserviceName).getShortName();

	appManager.getEventBus().post(new CreateMicroserviceEvent(this));

	LOGGER.info("create MicroserviceVersions, appId={}, microserviceName={}.",
	appId,
	microserviceName);
	}

	public boolean isWaitingDelete() {
	return waitingDelete;
	}

	public MicroserviceVersions markWaitingDelete() {
	this.waitingDelete = true;
	return this;
	}

	public AppManager getAppManager() {
	return appManager;
	}

	public String getAppId() {
	return appId;
	}

	public String getMicroserviceName() {
	return microserviceName;
	}

	public String getShortName() {
	return shortName;
	}

	public Map<String, MicroserviceVersion> getVersions() {
	return versions;
	}

	@SuppressWarnings("unchecked")
	public <T extends MicroserviceVersion> T getVersion(String serviceId) {
	return (T) versions.get(serviceId);
	}

	public String getRevision() {
	return revision;
	}

	public void setRevision(String revision) {
	this.revision = revision;
	}

	public List<MicroserviceInstance> getPulledInstances() {
	return pulledInstances;
	}

	public long getLastPullTime() {
	return lastPullTime;
	}

	public MicroserviceInstances getLastPulledResult() {
	return lastPulledResult;
	}

	public VendorExtensions getVendorExtensions() {
	return vendorExtensions;
	}

	public void pullInstances() {
	lastPullTime = System.currentTimeMillis();
	MicroserviceInstances microserviceInstances = findServiceInstances();
	lastPulledResult = microserviceInstances;
	if (microserviceInstances == null) {
	// pulled failed, did not get anything
	// will not do anything, consumers will use existing instances
	return;
	}
	if (microserviceInstances.isMicroserviceNotExist()) {
	// pulled failed, SC said target not exist
	waitingDelete = true;
	return;
	}

	if (null != revision && revision.equals(microserviceInstances.getRevision())) {
	return;
	}

	pulledInstances = microserviceInstances.getInstancesResponse().getInstances();
	pulledInstances.sort(Comparator.comparing(MicroserviceInstance::getInstanceId));
	String rev = microserviceInstances.getRevision();

	safeSetInstances(pulledInstances, rev);
	}

	protected MicroserviceInstances findServiceInstances() {
	return DiscoveryManager.INSTANCE.findServiceInstances(appId,
	microserviceName,
	DefinitionConst.VERSION_RULE_ALL,
	revision);
	}

	protected void safeSetInstances(List<MicroserviceInstance> pulledInstances, String rev) {
	try {
	List<MicroserviceInstance> filteredInstance = pulledInstances;
	// 增加一个配置项只使用 `UP` 实例。在使用 `TESTING` 进行拨测，并且配置了
	// servicecomb.references.version-rule=latest 场景，需要保证不使用
	// `TESTING` 实例。不能依赖 InstanceStatusDiscoveryFilter, 避免
	// 构建的 VersionRule 实例列表为空。
	if (ServiceRegistryCommonConfig.useUpInstancesOnly()) {
	filteredInstance = pulledInstances.stream().filter(item -> MicroserviceInstanceStatus.UP == item.getStatus())
	.collect(Collectors.toList());
	}
	setInstances(filteredInstance, rev);
	} catch (Throwable e) {
	waitingDelete = true;
	LOGGER.error("Failed to setInstances, appId={}, microserviceName={}.",
	getAppId(),
	getMicroserviceName(),
	e);
	}
	}

	static class MergedInstances {
	// key is microserviceId
	Map<String, List<MicroserviceInstance>> microserviceIdMap = new HashMap<>();

	// key is instanceId
	Map<String, MicroserviceInstance> instanceIdMap = new HashMap<>();

	void addInstance(MicroserviceInstance instance) {
	instanceIdMap.put(instance.getInstanceId(), instance);
	microserviceIdMap
	.computeIfAbsent(instance.getServiceId(), key -> new ArrayList<>())
	.add(instance);
	}
	}

	private void setInstances(List<MicroserviceInstance> pulledInstances, String rev) {
	synchronized (lock) {
	MergedInstances mergedInstances = mergeInstances(pulledInstances, instances);
	instances = mergedInstances.instanceIdMap.values();
	// clear cache
	versions.forEach((key, value) -> value.setInstances(new ArrayList<>()));
	for (Entry<String, List<MicroserviceInstance>> entry : mergedInstances.microserviceIdMap.entrySet()) {
	// ensure microserviceVersion exists
	versions.computeIfAbsent(entry.getKey(),
	microserviceId -> createMicroserviceVersion(microserviceId, entry.getValue()))
	.setInstances(entry.getValue());
	}

	for (MicroserviceVersionRule microserviceVersionRule : versionRules.values()) {
	microserviceVersionRule.update(versions, instances);
	}
	revision = rev;
	}
	}

	protected MicroserviceVersion createMicroserviceVersion(String microserviceId, List<MicroserviceInstance> instances) {
	return new MicroserviceVersion(this, microserviceId, microserviceName, instances);
	}

	private MergedInstances mergeInstances(List<MicroserviceInstance> pulledInstances,
	Collection<MicroserviceInstance> inUseInstances) {
	MergedInstances mergedInstances = new MergedInstances();
	pulledInstances.stream().forEach(mergedInstances::addInstance);
	MicroserviceInstancePing ping = SPIServiceUtils.getPriorityHighestService(MicroserviceInstancePing.class);

	if (pulledInstances.isEmpty() && inUseInstances != null && ServiceRegistryCommonConfig
	.isEmptyInstanceProtectionEnabled()) {

	inUseInstances.stream().forEach(instance -> {
	if (!mergedInstances.instanceIdMap.containsKey(instance.getInstanceId())) {
	if (ping.ping(instance)) {
	mergedInstances.addInstance(instance);
	}
	}
	});
	}

	return mergedInstances;
	}

	public MicroserviceVersionRule getOrCreateMicroserviceVersionRule(String versionRule) {
	// do not use computeIfAbsent
	MicroserviceVersionRule microserviceVersionRule = versionRules.get(versionRule);
	if (microserviceVersionRule == null) {
	synchronized (lock) {
	microserviceVersionRule = versionRules.computeIfAbsent(versionRule, this::createAndInitMicroserviceVersionRule);
	}
	}

	return microserviceVersionRule;
	}

	protected MicroserviceVersionRule createAndInitMicroserviceVersionRule(String strVersionRule) {
	LOGGER.info("create MicroserviceVersionRule, appId={}, microserviceName={}, versionRule={}.",
	appId,
	microserviceName,
	strVersionRule);

	MicroserviceVersionRule microserviceVersionRule =
	new MicroserviceVersionRule(appId, microserviceName, strVersionRule);
	microserviceVersionRule.update(versions, instances);
	return microserviceVersionRule;
	}

	public void onMicroserviceInstanceChanged(MicroserviceInstanceChangedEvent changedEvent) {
	if (!isEventAccept(changedEvent)) {
	return;
	}
	// pull instances always replace old instances, not append
	//
	// pull result and watch event sequence is not defined even inside SC.
	// it's not safe to trust the event, so we just send a new pull request
	//
	// CREATE/UPDATE:
	// if pull 1/2/3, and then add 4, but "add 4" received before pull result, will lost 4.
	// DELETE:
	// if pull 1/2/3, and then delete 3, but "delete 3" received before pull result, will have wrong 3.
	// EXPIRE::
	// black/white config in SC changed, we must refresh all data from sc.
	pullInstances();
	}

	protected boolean isEventAccept(MicroserviceInstanceChangedEvent changedEvent) {
	return (appId.equals(changedEvent.getKey().getAppId()) &&
	microserviceName.equals(changedEvent.getKey().getServiceName())) \|\|
	microserviceName.equals(
	changedEvent.getKey().getAppId() + DefinitionConst.APP_SERVICE_SEPARATOR + changedEvent.getKey()
	.getServiceName());
	}

	public void destroy() {
	for (MicroserviceVersion microserviceVersion : versions.values()) {
	microserviceVersion.destroy();
	}

	appManager.getEventBus().post(new DestroyMicroserviceEvent(this));
	}
	}