Soul Gateway source code analysis (16) Upstream service exploration

[soul-admin] [upstreams] [soul-admin] [upstreams] [upstreams] [soul-admin] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] [upstreams] Let’s dig into the source code to see how it works:

Soul-admin UpstreamCheckService#setup() is initialized from db after loading into upstreamMap.

/**
 * this is divide  http url upstream.
 *
 * @author xiaoyu
 */
@Slf4j
@Component
public class UpstreamCheckService {

	"// use upstream
    private static final Map<String, List<DivideUpstream>> UPSTREAM_MAP = Maps.newConcurrentMap();

	// Whether to turn on the probe switch
    @Value("${soul.upstream.check:true}")
    private boolean check;
	
    // The duration of periodic probing is 10 seconds by default
    @Value("${soul.upstream.scheduledTime:10}")
    private int scheduledTime;

    private final SelectorMapper selectorMapper;

    private final ApplicationEventPublisher eventPublisher;

    private final PluginMapper pluginMapper;

    private final SelectorConditionMapper selectorConditionMapper;
    
    / /...
    
    /** * load Selector configuration */
    @PostConstruct
    public void setup(a) {
        PluginDO pluginDO = pluginMapper.selectByName(PluginEnum.DIVIDE.getName());
        if(pluginDO ! =null) {
            List<SelectorDO> selectorDOList = selectorMapper.findByPluginId(pluginDO.getId());
            // Iterate over the Selector record and place upstream in the corresponding Selector name
            for (SelectorDO selectorDO : selectorDOList) {
                List<DivideUpstream> divideUpstreams = GsonUtils.getInstance().fromList(selectorDO.getHandle(), DivideUpstream.class);
                if(CollectionUtils.isNotEmpty(divideUpstreams)) { UPSTREAM_MAP.put(selectorDO.getName(), divideUpstreams); }}}if (check) {
            new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), SoulThreadFactory.create("scheduled-upstream-task".false))
                    .scheduleWithFixedDelay(this::scheduled, 10, scheduledTime, TimeUnit.SECONDS); }}/ /...
    
    /**
     * Submit.
     *
     * @param selectorName   the selector name
     * @param divideUpstream the divide upstream
     */
    public void submit(final String selectorName, final DivideUpstream divideUpstream) {
        if (UPSTREAM_MAP.containsKey(selectorName)) {
            UPSTREAM_MAP.get(selectorName).add(divideUpstream);
        } else{ UPSTREAM_MAP.put(selectorName, Lists.newArrayList(divideUpstream)); }}/**
     * Replace.
     *
     * @param selectorName    the selector name
     * @param divideUpstreams the divide upstream list
     */
    public void replace(final String selectorName, final List<DivideUpstream> divideUpstreams) {
        UPSTREAM_MAP.put(selectorName, divideUpstreams);
    }

	// Use a timed thread pool for probing
    private void scheduled(a) {
        if (UPSTREAM_MAP.size() > 0) {
            UPSTREAM_MAP.forEach(this::check); }}// Core method to perform checkUrl on Upstream in each selector
    private void check(final String selectorName, final List<DivideUpstream> upstreamList) {
        List<DivideUpstream> successList = Lists.newArrayListWithCapacity(upstreamList.size());
        for (DivideUpstream divideUpstream : upstreamList) {
            final boolean pass = UpstreamCheckUtils.checkUrl(divideUpstream.getUpstreamUrl());
            if (pass) {
                if(! divideUpstream.isStatus()) { divideUpstream.setTimestamp(System.currentTimeMillis()); divideUpstream.setStatus(true);
                    log.info("UpstreamCacheManager check success the url: {}, host: {} ", divideUpstream.getUpstreamUrl(), divideUpstream.getUpstreamHost());
                }
                successList.add(divideUpstream);
            } else {
                divideUpstream.setStatus(false);
                log.error("check the url={} is fail ", divideUpstream.getUpstreamUrl()); }}if (successList.size() == upstreamList.size()) {
            return;
        }
        if (successList.size() > 0) {
            UPSTREAM_MAP.put(selectorName, successList);
            updateSelectorHandler(selectorName, successList);
        } else {
            UPSTREAM_MAP.remove(selectorName);
            updateSelectorHandler(selectorName, null); }}/ /...
}
Copy the code

Here is UpstreamCheckUtils:

public class UpstreamCheckUtils {

	// Regular expression to filter out valid IP
    private static final Pattern PATTERN = Pattern
            .compile("(http://|https://)? (? : (? : [0, 1]? \\d? \\d|2[0-4]\\d|25[0-5])\\.) {3} (? : [0, 1]? \\d? \ \ | 2 d \ \ [0-4] d | 25 [0-5]) : \ \ d {0, 5}");

    private static final String HTTP = "http";

    /**
     * Check url boolean.
     *
     * @param url the url
     * @return the boolean
     */
    public static boolean checkUrl(final String url) {
        if (StringUtils.isBlank(url)) {
            return false;
        }
        if (checkIP(url)) {
            String[] hostPort;
            // If HTTP is used
            if (url.startsWith(HTTP)) {
                final String[] http = StringUtils.split(url, "\ \ \ \ /");
                hostPort = StringUtils.split(http[1], Constants.COLONS);
            } else {
                hostPort = StringUtils.split(url, Constants.COLONS);
            }
            return isHostConnector(hostPort[0], Integer.parseInt(hostPort[1]));
        } else {
            returnisHostReachable(url); }}private static boolean checkIP(final String url) {
        return PATTERN.matcher(url).matches();
    }

    private static boolean isHostConnector(final String host, final int port) {
        try (Socket socket = new Socket()) {
        	// use the socket client to connect to upstream
            socket.connect(new InetSocketAddress(host, port));
        } catch (IOException e) {
        	// If an exception is thrown, false is returned
            return false;
        }
        return true;
    }
	
    // Check whether the domain name is reachable
    private static boolean isHostReachable(final String host) {
        try {
            return InetAddress.getByName(host).isReachable(1000);
        } catch (IOException ignored) {
        }
        return false; }}Copy the code