samza-core/src/main/java/org/apache/samza/coordinator/StreamRegexMonitor.java - samza - 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.samza.coordinator;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import org.apache.samza.metrics.Gauge;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.system.StreamMetadataCache;
 import org.apache.samza.system.SystemStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.collection.JavaConverters;


 /**
  * A single-thread based monitor that periodically monitors the given set of stream regexes, and matches them to
  * the given set of streams. If a stream matching a given regex that is not in the corresponding stream set is detected,
  * it invokes a {@link StreamRegexMonitor.Callback} with the initial input set, the new input stream set, and the regexes
  * being monitored.
  */
 public class StreamRegexMonitor {
   private static final Logger log = LoggerFactory.getLogger(StreamRegexMonitor.class);

   // Factory of daemon-threads to create the single threaded executor pool
   private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder().setDaemon(true)
       .setNameFormat("Samza-" + StreamRegexMonitor.class.getSimpleName())
       .build();

   // Enum to describe the state of the regexMonitor
   private enum State {
     INIT, RUNNING, STOPPED
   }

   private final Set<SystemStream> streamsToMonitor;
   private final Map<String, Pattern> systemRegexesToMonitor;
   private final StreamMetadataCache metadataCache;
   private final int inputRegexMonitorPeriodMs;

   // Map of gauges (one per system), emitted when new input stream for that system is detected
   private final Map<String, Gauge<Integer>> gauges;

   private final Callback callbackMethod;

   // Used to guard write access to state.
   private final Object lock = new Object();

   private final ScheduledExecutorService schedulerService = Executors.newSingleThreadScheduledExecutor(THREAD_FACTORY);

   private volatile State state = State.INIT;

   /**
    * A callback that is invoked when the {@link StreamRegexMonitor} detects a new input stream matching given regex.
    */
   public interface Callback {
     /**
      * Method to be called when new input streams are detected.
      * @param initialInputSet The initial set of input streams
      * @param newInputStreams The set of new input streams discovered
      * @param regexesMonitored The set of regexes being monitored
      */
     void onInputStreamsChanged(Set<SystemStream> initialInputSet, Set<SystemStream> newInputStreams,
         Map<String, Pattern> regexesMonitored);
   }

   /**
    * Default constructor.
    *
    *  @param streamsToMonitor  a set of SystemStreams to monitor
    * @param systemRegexesToMonitor  map of regexes for each input system
    * @param metadataCache     the metadata cache which will be used to fetch metadata for partition counts.
    * @param metrics           the metrics registry to which the metrics should be added.
    * @param inputRegexMonitorPeriodMs the period at which the monitor will check each input-regex
    * @param monitorCallback   the callback method to be invoked when new input stream matching regex is detected
    */
   public StreamRegexMonitor(Set<SystemStream> streamsToMonitor, Map<String, Pattern> systemRegexesToMonitor,
       StreamMetadataCache metadataCache, MetricsRegistry metrics, int inputRegexMonitorPeriodMs,
       Callback monitorCallback) {
     this.streamsToMonitor = streamsToMonitor;
     this.systemRegexesToMonitor = systemRegexesToMonitor;
     this.metadataCache = metadataCache;
     this.callbackMethod = monitorCallback;
     this.inputRegexMonitorPeriodMs = inputRegexMonitorPeriodMs;

     // Pre-populate the gauges
     Map<String, Gauge<Integer>> mutableGauges = new HashMap<>();
     for (String systemToMonitor : systemRegexesToMonitor.keySet()) {
       Gauge gauge = metrics.newGauge("job-coordinator", String.format("%s-new-input-streams", systemToMonitor), 0);
       mutableGauges.put(systemToMonitor, gauge);
     }
     gauges = Collections.unmodifiableMap(mutableGauges);

     log.info("Created {} with inputRegexMonitorPeriodMs: {} and systemRegexesToMonitor: {}", this.getClass().getName(),
         this.inputRegexMonitorPeriodMs, this.systemRegexesToMonitor);
   }

   /**
    * Starts the monitor.
    */
   public void start() {
     synchronized (lock) {
       switch (state) {
         case INIT:
           if (inputRegexMonitorPeriodMs > 0) {
             schedulerService.scheduleAtFixedRate(new Runnable() {
               @Override
               public void run() {
                 monitorInputRegexes();
               }
             }, 0, inputRegexMonitorPeriodMs, TimeUnit.MILLISECONDS);
           }
           state = State.RUNNING;
           break;

         case RUNNING:
           // start is idempotent
           return;

         case STOPPED:
           throw new IllegalStateException("StreamRegexMonitor was stopped and cannot be restarted.");
       }
     }
   }

   /**
    * Stops the monitor. Once it stops, it cannot be restarted.
    */
   public void stop() {
     synchronized (lock) {
       // We could also wait for full termination of the scheduler service, but it is overkill for
       // our use case.
       schedulerService.shutdownNow();

       state = State.STOPPED;
     }
   }

   private void monitorInputRegexes() {
     log.debug("Running monitorInputRegexes");

     try {
       // obtain the list of SysStreams that match given patterns for all systems
       Set<SystemStream> inputStreamsMatchingPattern = new HashSet<>();

       // For each input system, for which we have a regex to monitor
       for (String systemName : this.systemRegexesToMonitor.keySet()) {

         try {
           // obtain the list of SysStreams that match the regex for this system
           // using the systemAdmin in the metadataCache
           inputStreamsMatchingPattern.addAll(
               JavaConverters.setAsJavaSetConverter(this.metadataCache.getAllSystemStreams(systemName))
                   .asJava()
                   .stream()
                   .filter(x -> x.getStream().matches(this.systemRegexesToMonitor.get(systemName).pattern()))
                   .collect(Collectors.toSet()));
         } catch (UnsupportedOperationException e) {
           log.error("UnsupportedOperationException while monitoring input regexes for system {}", systemName, e);
         }
       }

       // if there is a stream that is in the input-Set but not in the streamsToMonitor
       // since streamsToMonitor = task.inputs
       if (!streamsToMonitor.containsAll(inputStreamsMatchingPattern)) {
         log.info("New input system-streams discovered. InputStreamsMatchingPattern: {} but streamsToMonitor: {} ",
             inputStreamsMatchingPattern, streamsToMonitor);

         // invoke notify callback with new input streams
         this.callbackMethod.onInputStreamsChanged(streamsToMonitor,
             Sets.difference(inputStreamsMatchingPattern, streamsToMonitor), systemRegexesToMonitor);
       } else {
         log.info("No new input system-Streams discovered streamsToMonitor {} inputStreamsMatchingPattern {}",
             streamsToMonitor, inputStreamsMatchingPattern);
       }
     } catch (Exception e) {
       log.error("Exception while monitoring input regexes.", e);
     }
   }

   @VisibleForTesting
   boolean isRunning() {
     return state == State.RUNNING;
   }

   /**
    * Wait until this service has shutdown. Returns true if shutdown occurred within the timeout
    * and false otherwise.
    * <p>
    * This is currently exposed at the package private level for tests only.
    */
   @VisibleForTesting
   boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
     return schedulerService.awaitTermination(timeout, unit);
   }
 }
	/*
	* 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.samza.coordinator;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.collect.Sets;
	import com.google.common.util.concurrent.ThreadFactoryBuilder;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.Executors;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.ThreadFactory;
	import java.util.concurrent.TimeUnit;
	import java.util.regex.Pattern;
	import java.util.stream.Collectors;
	import org.apache.samza.metrics.Gauge;
	import org.apache.samza.metrics.MetricsRegistry;
	import org.apache.samza.system.StreamMetadataCache;
	import org.apache.samza.system.SystemStream;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import scala.collection.JavaConverters;


	/**
	* A single-thread based monitor that periodically monitors the given set of stream regexes, and matches them to
	* the given set of streams. If a stream matching a given regex that is not in the corresponding stream set is detected,
	* it invokes a {@link StreamRegexMonitor.Callback} with the initial input set, the new input stream set, and the regexes
	* being monitored.
	*/
	public class StreamRegexMonitor {
	private static final Logger log = LoggerFactory.getLogger(StreamRegexMonitor.class);

	// Factory of daemon-threads to create the single threaded executor pool
	private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder().setDaemon(true)
	.setNameFormat("Samza-" + StreamRegexMonitor.class.getSimpleName())
	.build();

	// Enum to describe the state of the regexMonitor
	private enum State {
	INIT, RUNNING, STOPPED
	}

	private final Set<SystemStream> streamsToMonitor;
	private final Map<String, Pattern> systemRegexesToMonitor;
	private final StreamMetadataCache metadataCache;
	private final int inputRegexMonitorPeriodMs;

	// Map of gauges (one per system), emitted when new input stream for that system is detected
	private final Map<String, Gauge<Integer>> gauges;

	private final Callback callbackMethod;

	// Used to guard write access to state.
	private final Object lock = new Object();

	private final ScheduledExecutorService schedulerService = Executors.newSingleThreadScheduledExecutor(THREAD_FACTORY);

	private volatile State state = State.INIT;

	/**
	* A callback that is invoked when the {@link StreamRegexMonitor} detects a new input stream matching given regex.
	*/
	public interface Callback {
	/**
	* Method to be called when new input streams are detected.
	* @param initialInputSet The initial set of input streams
	* @param newInputStreams The set of new input streams discovered
	* @param regexesMonitored The set of regexes being monitored
	*/
	void onInputStreamsChanged(Set<SystemStream> initialInputSet, Set<SystemStream> newInputStreams,
	Map<String, Pattern> regexesMonitored);
	}

	/**
	* Default constructor.
	*
	* @param streamsToMonitor a set of SystemStreams to monitor
	* @param systemRegexesToMonitor map of regexes for each input system
	* @param metadataCache the metadata cache which will be used to fetch metadata for partition counts.
	* @param metrics the metrics registry to which the metrics should be added.
	* @param inputRegexMonitorPeriodMs the period at which the monitor will check each input-regex
	* @param monitorCallback the callback method to be invoked when new input stream matching regex is detected
	*/
	public StreamRegexMonitor(Set<SystemStream> streamsToMonitor, Map<String, Pattern> systemRegexesToMonitor,
	StreamMetadataCache metadataCache, MetricsRegistry metrics, int inputRegexMonitorPeriodMs,
	Callback monitorCallback) {
	this.streamsToMonitor = streamsToMonitor;
	this.systemRegexesToMonitor = systemRegexesToMonitor;
	this.metadataCache = metadataCache;
	this.callbackMethod = monitorCallback;
	this.inputRegexMonitorPeriodMs = inputRegexMonitorPeriodMs;

	// Pre-populate the gauges
	Map<String, Gauge<Integer>> mutableGauges = new HashMap<>();
	for (String systemToMonitor : systemRegexesToMonitor.keySet()) {
	Gauge gauge = metrics.newGauge("job-coordinator", String.format("%s-new-input-streams", systemToMonitor), 0);
	mutableGauges.put(systemToMonitor, gauge);
	}
	gauges = Collections.unmodifiableMap(mutableGauges);

	log.info("Created {} with inputRegexMonitorPeriodMs: {} and systemRegexesToMonitor: {}", this.getClass().getName(),
	this.inputRegexMonitorPeriodMs, this.systemRegexesToMonitor);
	}

	/**
	* Starts the monitor.
	*/
	public void start() {
	synchronized (lock) {
	switch (state) {
	case INIT:
	if (inputRegexMonitorPeriodMs > 0) {
	schedulerService.scheduleAtFixedRate(new Runnable() {
	@Override
	public void run() {
	monitorInputRegexes();
	}
	}, 0, inputRegexMonitorPeriodMs, TimeUnit.MILLISECONDS);
	}
	state = State.RUNNING;
	break;

	case RUNNING:
	// start is idempotent
	return;

	case STOPPED:
	throw new IllegalStateException("StreamRegexMonitor was stopped and cannot be restarted.");
	}
	}
	}

	/**
	* Stops the monitor. Once it stops, it cannot be restarted.
	*/
	public void stop() {
	synchronized (lock) {
	// We could also wait for full termination of the scheduler service, but it is overkill for
	// our use case.
	schedulerService.shutdownNow();

	state = State.STOPPED;
	}
	}

	private void monitorInputRegexes() {
	log.debug("Running monitorInputRegexes");

	try {
	// obtain the list of SysStreams that match given patterns for all systems
	Set<SystemStream> inputStreamsMatchingPattern = new HashSet<>();

	// For each input system, for which we have a regex to monitor
	for (String systemName : this.systemRegexesToMonitor.keySet()) {

	try {
	// obtain the list of SysStreams that match the regex for this system
	// using the systemAdmin in the metadataCache
	inputStreamsMatchingPattern.addAll(
	JavaConverters.setAsJavaSetConverter(this.metadataCache.getAllSystemStreams(systemName))
	.asJava()
	.stream()
	.filter(x -> x.getStream().matches(this.systemRegexesToMonitor.get(systemName).pattern()))
	.collect(Collectors.toSet()));
	} catch (UnsupportedOperationException e) {
	log.error("UnsupportedOperationException while monitoring input regexes for system {}", systemName, e);
	}
	}

	// if there is a stream that is in the input-Set but not in the streamsToMonitor
	// since streamsToMonitor = task.inputs
	if (!streamsToMonitor.containsAll(inputStreamsMatchingPattern)) {
	log.info("New input system-streams discovered. InputStreamsMatchingPattern: {} but streamsToMonitor: {} ",
	inputStreamsMatchingPattern, streamsToMonitor);

	// invoke notify callback with new input streams
	this.callbackMethod.onInputStreamsChanged(streamsToMonitor,
	Sets.difference(inputStreamsMatchingPattern, streamsToMonitor), systemRegexesToMonitor);
	} else {
	log.info("No new input system-Streams discovered streamsToMonitor {} inputStreamsMatchingPattern {}",
	streamsToMonitor, inputStreamsMatchingPattern);
	}
	} catch (Exception e) {
	log.error("Exception while monitoring input regexes.", e);
	}
	}

	@VisibleForTesting
	boolean isRunning() {
	return state == State.RUNNING;
	}

	/**
	* Wait until this service has shutdown. Returns true if shutdown occurred within the timeout
	* and false otherwise.
	* <p>
	* This is currently exposed at the package private level for tests only.
	*/
	@VisibleForTesting
	boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
	return schedulerService.awaitTermination(timeout, unit);
	}
	}