SaaS Crawler Module

.gitignore

@@ -16,3 +16,6 @@ data-prepper-main/src/test/resources/logstash-conf/logstash-filter.yaml
 
 # Python virtual environments
 .venv
+
+# output folder created when we run test cases
+**/out/

data-prepper-plugins/build.gradle

@@ -13,5 +13,5 @@ subprojects {
 }
 
 dependencies {
-    subprojects.forEach { api project(':data-prepper-plugins:' + it.name) }
+    subprojects.findAll { api project(it.path) }
 }

data-prepper-plugins/saas-source-plugins/build.gradle

@@ -0,0 +1,13 @@
+plugins {
+    id 'java-library'
+}
+
+
+subprojects {
+    apply plugin: 'data-prepper.publish'
+    group = 'org.opensearch.dataprepper.plugins.source'
+}
+
+dependencies {
+    subprojects.forEach { api project(':data-prepper-plugins::saas-source-plugins:' + it.name) }
+}

data-prepper-plugins/saas-source-plugins/jira-source/README.md

@@ -0,0 +1,8 @@
+
+# Metrics
+
+### Counter
+- `issuesRequested`: measures total number of issue Requests sent.
+
+### Timer
+- `requestProcessDuration`: measures latency of requests processed by the jira source plugin.

data-prepper-plugins/saas-source-plugins/jira-source/build.gradle

@@ -0,0 +1,31 @@
+plugins {
+    id 'java'
+}
+
+dependencies {
+
+    implementation project(path: ':data-prepper-plugins:saas-source-plugins:saas-crawler')
+    implementation project(path: ':data-prepper-api')
+    implementation project(path: ':data-prepper-plugins:aws-plugin-api')
+    implementation project(path: ':data-prepper-plugins:buffer-common')
+    implementation project(path: ':data-prepper-plugins:common')
+
+    implementation libs.commons.io
+    implementation 'io.micrometer:micrometer-core'
+    implementation 'com.fasterxml.jackson.core:jackson-core'
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation 'javax.inject:javax.inject:1'
+    implementation 'org.springframework:spring-web:5.3.39'
+    implementation 'org.springframework.retry:spring-retry:1.3.4'
+
+    implementation 'org.projectlombok:lombok:1.18.30'
+    annotationProcessor 'org.projectlombok:lombok:1.18.30'
+
+    implementation(libs.spring.context) {
+        exclude group: 'commons-logging', module: 'commons-logging'
+    }
+}
+
+test {
+    useJUnitPlatform()
+}

data-prepper-plugins/saas-source-plugins/jira-source/src/main/java/org/opensearch/dataprepper/plugins/source/saas/jira/JiraClient.java

@@ -0,0 +1,44 @@
+package org.opensearch.dataprepper.plugins.source.saas.jira;
+
+import org.opensearch.dataprepper.model.buffer.Buffer;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.base.SaasClient;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.base.SaasSourceConfig;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.coordination.state.SaasWorkerProgressState;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.model.ItemInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Named;
+import java.util.Iterator;
+
+/**
+ * This class represents a Jira client.
+ */
+@Named
+public class JiraClient implements SaasClient {
+
+    private static final Logger log = LoggerFactory.getLogger(JiraClient.class);
+    private long lastPollTime;
+
+    public JiraClient() {
+    }
+
+
+    @Override
+    public Iterator<ItemInfo> listItems() {
+        return null;
+    }
+
+    @Override
+    public void setLastPollTime(long lastPollTime) {
+        log.info("Setting the lastPollTime: {}", lastPollTime);
+        this.lastPollTime = lastPollTime;
+    }
+
+    @Override
+    public void executePartition(SaasWorkerProgressState state, Buffer<Record<Event>> buffer, SaasSourceConfig configuration) {
+        log.info("Logic for executing the partitions");
+    }
+}

data-prepper-plugins/saas-source-plugins/jira-source/src/main/java/org/opensearch/dataprepper/plugins/source/saas/jira/JiraSource.java

@@ -0,0 +1,57 @@
+package org.opensearch.dataprepper.plugins.source.saas.jira;
+
+
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.model.buffer.Buffer;
+import org.opensearch.dataprepper.model.codec.ByteDecoder;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.plugin.PluginFactory;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.source.Source;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.SaasCrawlerApplicationContextMarker;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.base.Crawler;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.base.SaasPluginExecutorServiceProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * JiraConnector connector entry point.
+ */
+
+@DataPrepperPlugin(name = "jira",
+        pluginType = Source.class,
+        packagesToScan = {SaasCrawlerApplicationContextMarker.class, JiraSource.class}
+)
+public class JiraSource implements Source<Record<Event>> {
+
+  private static final Logger log = LoggerFactory.getLogger(JiraSource.class);
+
+
+  @DataPrepperPluginConstructor
+  public JiraSource(final PluginMetrics pluginMetrics,
+                    final PluginFactory pluginFactory,
+                    final AcknowledgementSetManager acknowledgementSetManager,
+                    Crawler crawler,
+                    SaasPluginExecutorServiceProvider executorServiceProvider) {
+    log.info("Create Jira Source Connector");
+  }
+
+  public void start(Buffer<Record<Event>> buffer) {
+    log.info("Starting Jira Source Plugin... ");
+  }
+
+  @Override
+  public void stop() {
+
+  }
+
+  @Override
+  public ByteDecoder getDecoder() {
+    return Source.super.getDecoder();
+  }
+
+}

data-prepper-plugins/saas-source-plugins/saas-crawler/build.gradle

@@ -0,0 +1,30 @@
+plugins {
+    id 'java-library'
+}
+
+group = 'org.opensearch.dataprepper.plugins.source.saas_crawler'
+
+tasks.withType(Javadoc).configureEach {
+    enabled = false
+}
+
+dependencies {
+
+    implementation project(path: ':data-prepper-api')
+
+    implementation 'com.fasterxml.jackson.core:jackson-core'
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation 'org.projectlombok:lombok:1.18.30'
+    implementation 'javax.inject:javax.inject:1'
+    implementation 'javax.annotation:javax.annotation-api:1.3.2'
+
+    implementation(libs.spring.context) {
+        exclude group: 'commons-logging', module: 'commons-logging'
+    }
+
+    annotationProcessor 'org.projectlombok:lombok:1.18.30'
+}
+
+test {
+    useJUnitPlatform()
+}

data-prepper-plugins/saas-source-plugins/saas-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/saas_crawler/SaasCrawlerApplicationContextMarker.java

@@ -0,0 +1,7 @@
+package org.opensearch.dataprepper.plugins.source.saas_crawler;
+
+/**
+ * Market interface to indicate the base package to scan for dependency injection
+ */
+public interface SaasCrawlerApplicationContextMarker {
+}

data-prepper-plugins/saas-source-plugins/saas-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/saas_crawler/base/Crawler.java

@@ -0,0 +1,81 @@
+package org.opensearch.dataprepper.plugins.source.saas_crawler.base;
+
+import org.opensearch.dataprepper.model.buffer.Buffer;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourceCoordinator;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.coordination.partition.SaasSourcePartition;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.coordination.state.SaasWorkerProgressState;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.model.ItemInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Named;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Named
+public class Crawler {
+    private static final Logger log = LoggerFactory.getLogger(Crawler.class);
+    private static final int maxItemsPerPage = 20;
+    public static final String CREATED = "created";
+    public static final String UPDATED = "updated";
+
+    private final SaasClient client;
+
+    public Crawler(SaasClient client) {
+        this.client = client;
+    }
+
+    public long crawl(long lastPollTime,
+                      EnhancedSourceCoordinator coordinator) {
+        long startTime = System.currentTimeMillis();
+        client.setLastPollTime(lastPollTime);
+        Iterator<ItemInfo> itemInfoIterator = client.listItems();
+        log.info("Starting to crawl the source with lastPollTime: {}", lastPollTime);
+        long updatedPollTime = 0;
+        do {
+            final List<ItemInfo> itemInfoList = new ArrayList<>();
+            for (int i = 0; i < maxItemsPerPage && itemInfoIterator.hasNext(); i++) {
+                ItemInfo nextItem = itemInfoIterator.next();
+                if(nextItem==null) {
+                    //we don't expect null items, but just in case, we'll skip them
+                    continue;
+                }
+                itemInfoList.add(nextItem);
+                Map<String, Object> metadata = nextItem.getMetadata();
+                long niCreated = Long.parseLong(metadata.get(CREATED)!=null? (String)metadata.get(CREATED):"0");
+                long niUpdated = Long.parseLong(metadata.get(UPDATED)!=null? (String)metadata.get(UPDATED):"0");
+                updatedPollTime = Math.max(updatedPollTime, niCreated);
+                updatedPollTime = Math.max(updatedPollTime, niUpdated);
+            }
+            createPartition(itemInfoList, coordinator);
+        }while (itemInfoIterator.hasNext());
+        log.debug("Crawling completed in {} ms", System.currentTimeMillis() - startTime);
+        return updatedPollTime != 0 ? updatedPollTime : startTime;
+    }
+
+    public void executePartition(SaasWorkerProgressState state, Buffer<Record<Event>> buffer, SaasSourceConfig sourceConfig) {
+        client.executePartition(state, buffer, sourceConfig);
+    }
+
+    private void createPartition(List<ItemInfo> itemInfoList, EnhancedSourceCoordinator coordinator) {
+        if(itemInfoList.isEmpty()) {
+            return;
+        }
+        ItemInfo itemInfo = itemInfoList.get(0);
+        String partitionKey = itemInfo.getPartitionKey();
+        List<String> itemIds = itemInfoList.stream().map(ItemInfo::getId).collect(Collectors.toList());
+        SaasWorkerProgressState state = new SaasWorkerProgressState();
+        state.setKeyAttributes(itemInfo.getKeyAttributes());
+        state.setItemIds(itemIds);
+        state.setExportStartTime(System.currentTimeMillis());
+        state.setLoadedItems(itemInfoList.size());
+        SaasSourcePartition sourcePartition = new SaasSourcePartition(state, partitionKey);
+        coordinator.createPartition(sourcePartition);
+    }
+
+}

data-prepper-plugins/saas-source-plugins/saas-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/saas_crawler/base/SaasClient.java

@@ -0,0 +1,43 @@
+package org.opensearch.dataprepper.plugins.source.saas_crawler.base;
+
+import org.opensearch.dataprepper.model.buffer.Buffer;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.coordination.state.SaasWorkerProgressState;
+import org.opensearch.dataprepper.plugins.source.saas_crawler.model.ItemInfo;
+
+import java.util.Iterator;
+
+/**
+ * Interface for saas client. This interface can be implemented by different saas clients.
+ * For example, Jira, Salesforce, ServiceNow, etc.
+ */
+public interface SaasClient {
+
+
+    /**
+     * This will be the main API called by crawler. This method assumes that {@link
+     * SaasSourceConfig} is available as a member to {@link SaasClient}, as a result of
+     * which, other scanning properties will also be available to this method
+     *
+     * @return returns an {@link Iterator} of {@link ItemInfo}
+     */
+    Iterator<ItemInfo> listItems();
+
+
+    /**
+     * Method to set the last time we polled the service to check for any changes.
+     *
+     * @param lastPollTime time in milliseconds
+     */
+    void setLastPollTime(long lastPollTime);
+
+    /**
+     * Method for executing a particular partition or a chunk of work
+     *
+     * @param state worker node state holds the details of this particular chunk of work
+     * @param buffer pipeline buffer to write the results into
+     * @param sourceConfig pipeline configuration from the yaml
+     */
+    void executePartition(SaasWorkerProgressState state, Buffer<Record<Event>> buffer, SaasSourceConfig sourceConfig);
+}

data-prepper-plugins/saas-source-plugins/saas-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/saas_crawler/base/SaasPluginExecutorServiceProvider.java

@@ -0,0 +1,48 @@
+package org.opensearch.dataprepper.plugins.source.saas_crawler.base;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.PreDestroy;
+import javax.inject.Named;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+@Named
+public class SaasPluginExecutorServiceProvider {
+    private static final Logger  log = LoggerFactory.getLogger(SaasPluginExecutorServiceProvider.class);
+    public static final int DEFAULT_THREAD_COUNT = 50;
+    private final ExecutorService executorService;
+
+    public SaasPluginExecutorServiceProvider() {
+        executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_COUNT);
+    }
+
+    /**
+     * Constructor for testing
+     */
+    public SaasPluginExecutorServiceProvider(ExecutorService testExecutorService) {
+        executorService = testExecutorService;
+    }
+
+    public ExecutorService get() {
+        return executorService;
+    }
+
+    @PreDestroy
+    public void terminateExecutor() {
+        try {
+            log.debug("Shutting down ExecutorService " + executorService);
+            executorService.shutdown();
+            boolean isExecutorTerminated = executorService
+                    .awaitTermination(30, TimeUnit.SECONDS);
+            log.debug("ExecutorService terminated : " + isExecutorTerminated);
+        } catch (InterruptedException e) {
+            log.error("Interrupted while terminating executor : " + e.getMessage());
+            Thread.currentThread().interrupt();
+        } finally {
+            executorService.shutdownNow();
+        }
+    }
+}

data-prepper-plugins/saas-source-plugins/saas-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/saas_crawler/base/SaasSourceConfig.java

@@ -0,0 +1,9 @@
+package org.opensearch.dataprepper.plugins.source.saas_crawler.base;
+
+/**
+ * Marker interface to all the SAAS connectors configuration
+ */
+public interface SaasSourceConfig {
+
+    int DEFAULT_NUMBER_OF_WORKERS = 1;
+}