Watchdogv1

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/DeviceDriver.java

@@ -39,7 +39,7 @@ public DeviceDriver(DeviceDriverConfig config) {
     public String getProperty(String key) {
         final String value = config.getProperties().get(key);
         if (value == null || value.isEmpty()) {
-            throw new IllegalArgumentException(MessageFormat.format("Missing required parameter {0}", key));
+            throw new IllegalArgumentException(MessageFormat.format("PSC Service Error: Missing required parameter {0} in config", key));
         }
         return value;
     }

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/file/FileConfig.java

@@ -33,11 +33,13 @@ public class FileConfig {
 
     public final boolean enableLargeEvent;
 
+    public final String pscId;
+
 
     public FileConfig(String stateDatabaseFileName, String fileSpec, String fileExtension, String routingKey,
                       String streamName, String eventTemplateStr, int maxRecordsPerEvent, boolean enableDeleteCompletedFiles,
                       boolean exactlyOnce, double transactionTimeoutMinutes, long minTimeInMillisToUpdateFile, String fileType,
-                      boolean enableLargeEvent) {
+                      boolean enableLargeEvent, String pscId) {
         this.stateDatabaseFileName = stateDatabaseFileName;
         this.fileSpec = fileSpec;
         this.fileExtension = fileExtension;
@@ -51,6 +53,7 @@ public FileConfig(String stateDatabaseFileName, String fileSpec, String fileExte
         this.minTimeInMillisToUpdateFile = minTimeInMillisToUpdateFile;
         this.fileType = fileType;
         this.enableLargeEvent = enableLargeEvent;
+        this.pscId = pscId;
     }
 
     @Override
@@ -69,6 +72,7 @@ public String toString() {
                 ", transactionTimeoutMinutes=" + transactionTimeoutMinutes +
                 ", minTimeInMillisToUpdateFile=" + minTimeInMillisToUpdateFile +
                 ", enableLargeEvent=" + enableLargeEvent +
+                ", pscId=" + pscId +
                 '}';
     }
 }

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/file/FileIngestService.java

@@ -52,6 +52,7 @@ public abstract class FileIngestService extends DeviceDriver {
     private static final int DEFAULT_SAMPLES_PER_EVENT_KEY = 100;
 
     private static final int DEFAULT_INTERVAL_MS_KEY = 10000;
+    private static final String PSC_ID = "PSC_ID";
     private final FileProcessor processor;
     private final MetricPublisher metricPublisher;
     private final ScheduledExecutorService executor;
@@ -75,7 +76,8 @@ public FileIngestService(DeviceDriverConfig config) {
                 getTransactionTimeoutMinutes(),
                 getMinTimeInMillisToUpdateFile(),
                 config.getClassName(),
-                getLargeEventEnable());
+                getLargeEventEnable(),
+                getPscId());
         LOG.info("File Ingest Config: {}", fileSequenceConfig);
         final String scopeName = getScopeName();
         LOG.info("Scope: {}", scopeName);
@@ -88,12 +90,16 @@ public FileIngestService(DeviceDriverConfig config) {
         executor = Executors.newScheduledThreadPool(1, namedThreadFactory);
     }
 
+    private String getPscId() {
+        return getProperty(PSC_ID);
+    }
+
     String getFileSpec() {
         return getProperty(FILE_SPEC_KEY);
     }
 
     String getFileExtension() {
-        return getProperty(FILE_EXT, "");
+        return getProperty(FILE_EXT);
     }
 
     boolean getDeleteCompletedFiles() {

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/metrics/MetricsStore.java

@@ -38,6 +38,7 @@ public static Metric getMetric(String metricName) {
      * json string.
      */
     public static String getMetricsAsJson() throws JsonProcessingException {
+
         return MAPPER.writerWithDefaultPrettyPrinter()
                 .writeValueAsString(metricStore);
     }

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/metrics/PravegaClient.java

@@ -52,6 +52,7 @@ private EventStreamWriter<String> initializeWriter() {
         log.info("Initializing writer with {} {} {}", this.scope, this.streamName, this.controllerURI.toString());
         ClientConfig clientConfig = ClientConfig.builder().controllerURI(this.controllerURI).build();
         try (StreamManager streamManager = StreamManager.create(clientConfig)) {
+
             StreamConfiguration streamConfig = StreamConfiguration.builder()
                     .scalingPolicy(ScalingPolicy.fixed(1))
                     .build();

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/watchdog/PscWatchdogApp.java

@@ -28,6 +28,7 @@ public static void main(String[] args) {
             WatchDogConfig config = new WatchDogConfig(properties);
             log.debug("Properties: {}", properties);
             final WatchDogService service = new WatchDogService(config);
+            service.checkPscServiceStatus();
             service.startAsync();
             service.awaitTerminated();
         } catch (Exception e) {

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/watchdog/WatchDogConfig.java

@@ -12,6 +12,7 @@
 import io.pravega.common.util.Property;
 
 import java.io.File;
+import java.text.MessageFormat;
 import java.util.Map;
 
 public class WatchDogConfig {
@@ -32,7 +33,7 @@ public class WatchDogConfig {
     private final String serviceName;
 
     public WatchDogConfig(Map<String, String> properties) {
-        this.serviceName = properties.getOrDefault(PSC_SERVICE_NAME.toString(), PSC_SERVICE_NAME.getDefaultValue());
+        this.serviceName = getProperty(properties, PSC_SERVICE_NAME.toString());
         this.watchDogWatchIntervalSeconds = Integer.parseInt(properties.getOrDefault(PSC_WATCHDOG_WATCH_INTERVAL_SECONDS.toString(), PSC_WATCHDOG_WATCH_INTERVAL_SECONDS.getDefaultValue()));
         this.watchdogFileMonitorPath = properties.getOrDefault(PSC_WATCHDOG_FILE_MONITOR_PATH.toString(), PSC_WATCHDOG_FILE_MONITOR_PATH.getDefaultValue());
         this.restartTriggerPath = properties.getOrDefault(PSC_WATCHDOG_RESTART_TRIGGER_PATH.toString(), PSC_WATCHDOG_RESTART_TRIGGER_PATH.getDefaultValue());
@@ -79,4 +80,19 @@ public String getRestartTriggerPath() {
         return restartTriggerPath;
     }
 
+    /**
+     * Retrieves the value of a property from the given map of properties.
+     *
+     * @param  properties  the map of properties to retrieve the value from
+     * @param  key         the key of the property to retrieve
+     * @return             the value of the property
+     * @throws IllegalArgumentException throw IllegalArgumentException if there is missing required parameter.
+     */
+    public String getProperty(Map<String, String> properties, String key) {
+        final String value = properties.get(key);
+        if (value == null || value.isEmpty()) {
+            throw new IllegalArgumentException(MessageFormat.format("PSC Watchdog Service: Missing required parameter {0} in config", key));
+        }
+        return value;
+    }
 }

pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/watchdog/WatchDogService.java

@@ -10,9 +10,13 @@
 package io.pravega.sensor.collector.watchdog;
 
 import com.google.common.util.concurrent.AbstractService;
+import org.apache.commons.lang3.SystemUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
 
 /**
  * Watchdog service for Pravega Sensor Collector(PSC).
@@ -48,4 +52,28 @@ protected void doStop() {
         notifyStopped();
     }
 
+    public void checkPscServiceStatus() throws IOException {
+        Process psc = null;
+        log.info("Checking PSC service status {} ", this.config.getServiceName());
+        if (SystemUtils.IS_OS_LINUX) {
+            psc = Runtime.getRuntime().exec(new String[]{"sh", "-c", "systemctl status " + this.config.getServiceName()});
+        } else if ( SystemUtils.IS_OS_WINDOWS) {
+            psc = Runtime.getRuntime().exec(new String[]{"cmd.exe", "/c", this.config.getServiceName() + ".exe", "status"});
+        }
+        BufferedReader stdInput = new BufferedReader(new
+                InputStreamReader(psc.getInputStream()));
+
+        String s;
+        Boolean isAlive = true;
+        while ((s = stdInput.readLine()) != null) {
+            if (s.equalsIgnoreCase("NonExistent")) {
+                isAlive = false;
+            }
+        }
+        log.debug("Process psc {}, and isAlive value is {} ", psc, isAlive);
+        if (!isAlive) {
+            log.error("PSC service is not running");
+            throw new RuntimeException("PSC service is not running. Please start psc service before starting watchdog service.");
+        }
+    }
 }

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/file/FileIngestServiceTest.java

@@ -13,10 +13,14 @@
 import io.pravega.sensor.collector.DeviceDriverManager;
 import io.pravega.sensor.collector.Parameters;
 import io.pravega.sensor.collector.util.TestUtils;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.Map;
 
@@ -36,13 +40,24 @@ public class FileIngestServiceTest {
     private DeviceDriverManager driverManager;
 
     @BeforeEach
-    void setUp() {
+    void setUp() throws IOException {
         properties = Parameters.getProperties(FILE_NAME);
+        Files.deleteIfExists(Paths.get(properties.get("PRAVEGA_SENSOR_COLLECTOR_RAW1_DATABASE_FILE")));
         driverManager = new DeviceDriverManager(properties);
         deviceDriverConfig = new DeviceDriverConfig("RAW1", "RawFileIngestService",
                 TestUtils.configFromProperties(PREFIX, SEPARATOR, properties), driverManager);
     }
 
+    @AfterEach
+    public void tearDown() {
+        try {
+            Files.deleteIfExists(Paths.get(properties.get("PRAVEGA_SENSOR_COLLECTOR_RAW1_DATABASE_FILE")));
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        properties = null;
+    }
+
     @Test
     public void testFileIngestService() {
         FileIngestService fileIngestService = new MockFileIngestService(deviceDriverConfig);

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/file/FileProcessorFactoryTest.java

@@ -46,11 +46,9 @@ public void createRAWFileProcessorTest() throws Exception {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig(stateDatabaseFileName, "/opt/pravega-sensor-collector/Files/A", "parquet", "key12",
                 "stream1", "{}", 10, false,
-                true, 20.0, 5000L, "RawFileIngestService", true);
+                true, 20.0, 5000L, "RawFileIngestService", true, "psc1");
         FileProcessor rawFileProcessor = FileProcessorFactory.createFileSequenceProcessor(config, state, writer, transactionCoordinator, "writerId");
-
         Assertions.assertTrue(rawFileProcessor instanceof RawFileProcessor);
-
     }
 
     /*
@@ -61,9 +59,8 @@ public void createCSVFileProcessorTest() throws Exception {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig(stateDatabaseFileName, "/opt/pravega-sensor-collector/Files/A", "parquet", "key12",
                 "stream1", "{}", 10, false,
-                true, 20.0, 5000L, "CsvFileIngestService", false);
+                true, 20.0, 5000L, "CsvFileIngestService", false, "psc1");
         FileProcessor csvFileProcessor = FileProcessorFactory.createFileSequenceProcessor(config, state, writer, transactionCoordinator, "writerId");
-
         Assertions.assertTrue(csvFileProcessor instanceof CsvFileSequenceProcessor);
 
     }
@@ -76,7 +73,7 @@ public void createParquetFileProcessorTest() throws Exception {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig(stateDatabaseFileName, "/opt/pravega-sensor-collector/Files/A", "parquet", "key12",
                 "stream1", "{}", 10, false,
-                true, 20.0, 5000L, "ParquetFileIngestService", false);
+                true, 20.0, 5000L, "ParquetFileIngestService", false, "psc1");
         FileProcessor parquetFileProcessor = FileProcessorFactory.createFileSequenceProcessor(config, state, writer, transactionCoordinator, "writerId");
 
         Assertions.assertTrue(parquetFileProcessor instanceof ParquetFileProcessor);

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/file/FileProcessorTests.java

@@ -81,7 +81,7 @@ protected void setup() {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig("./psc.db", "/opt/pravega-sensor-collector/Files/A", "parquet", "key12",
                 "stream1", "{}", 10, true,
-                true, 20.0, 5000, "RawFileIngestService", true);
+                true, 20.0, 5000, "RawFileIngestService", true, "pscId");
     }
 
     @Test
@@ -225,7 +225,7 @@ public void testCreateRawFileProcessorWithNullTransactionCordinator() {
     public void testCreateRawFileProcessorWithNullStateDatabaseFilenameInConfig() {
         FileConfig newConfig = new FileConfig(null, "/opt/pravega-sensor-collector/Files/A", "parquet", "key12",
                 "stream1", "{}", 10, false,
-                true, 20.0, 5000, "RawFileIngestService", true);
+                true, 20.0, 5000, "RawFileIngestService", true, "pscId");
         Exception exception = Assert.assertThrows(NullPointerException.class, () -> new RawFileProcessor(newConfig, state, transactionalEventWriter, transactionCoordinator, "test"));
         Assert.assertTrue("config.stateDatabaseFileName".equals(exception.getMessage()));
     }

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/watchdog/WatchdogServiceTests.java

@@ -22,7 +22,7 @@ public class WatchdogServiceTests {
 
     @Test
     public void testWatchDogServiceStart() {
-        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().build());
+        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().put("PSC_SERVICE_NAME", "test").build());
         WatchDogService service = new WatchDogService(config);
         service.startAsync();
         Assert.assertTrue(service.isRunning());
@@ -32,7 +32,7 @@ public void testWatchDogServiceStart() {
 
     @Test
     public void testWatchdogMonitorStart() {
-        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().build());
+        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().put("PSC_SERVICE_NAME", "test").build());
         PSCWatchdogMonitor monitor = new PSCWatchdogMonitor(config);
         monitor.startAsync();
         Assert.assertTrue(monitor.isRunning());
@@ -42,7 +42,7 @@ public void testWatchdogMonitorStart() {
 
     @Test
     public void testWatchdogRestartPSC() {
-        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().build());
+        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().put("PSC_SERVICE_NAME", "test").build());
         PSCWatchdogMonitor monitor = new PSCWatchdogMonitor(config);
         monitor.onUpdateMissed();
         monitor.onUpdateMissed();
@@ -54,7 +54,7 @@ public void testWatchdogRestartPSC() {
 
     @Test
     public void testWatchdogResetCounterOnUpdates() {
-        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().build());
+        WatchDogConfig config = new WatchDogConfig(ImmutableMap.<String, String>builder().put("PSC_SERVICE_NAME", "test").build());
         PSCWatchdogMonitor monitor = new PSCWatchdogMonitor(config);
         monitor.onUpdateMissed();
         monitor.onUpdateMissed();

pravega-sensor-collector/src/test/resources/RawFileIngest-integration-test.properties

@@ -22,3 +22,4 @@ PRAVEGA_SENSOR_COLLECTOR_RAW1_TRANSACTION_TIMEOUT_MINUTES=2.0
 PRAVEGA_SENSOR_COLLECTOR_RAW1_CREATE_SCOPE=true
 PRAVEGA_SENSOR_COLLECTOR_RAW1_MIN_TIME_IN_MILLIS_TO_UPDATE_FILE=5000
 PRAVEGA_SENSOR_COLLECTOR_RAW1_DELETE_COMPLETED_FILES_INTERVAL_IN_SECONDS=15
+PRAVEGA_SENSOR_COLLECTOR_RAW1_PSC_ID=test

pravega-sensor-collector/src/test/resources/RawFileIngestService.properties

@@ -24,3 +24,4 @@ PRAVEGA_SENSOR_COLLECTOR_RAW1_TRANSACTION_TIMEOUT_MINUTES=1.0
 PRAVEGA_SENSOR_COLLECTOR_RAW1_CREATE_SCOPE=true
 PRAVEGA_SENSOR_COLLECTOR_RAW1_MIN_TIME_IN_MILLIS_TO_UPDATE_FILE=5000
 PRAVEGA_SENSOR_COLLECTOR_RAW1_DELETE_COMPLETED_FILES_INTERVAL_IN_SECONDS=15
+PRAVEGA_SENSOR_COLLECTOR_RAW1_PSC_ID=test