Create datasource API (opensearch-project#1458) (opensearch-project#1479

)

Signed-off-by: vamsi-amazon <reddyvam@amazon.com>

.gitignore

@@ -46,4 +46,5 @@ gen
 /.prom.pid.lock
 
 .java-version
-.worktrees
+.worktrees
+http-client.env.json

build.gradle

@@ -12,6 +12,7 @@ buildscript {
         version_tokens = opensearch_version.tokenize('-')
         opensearch_build = version_tokens[0] + '.0'
         prometheus_binary_version = "2.37.2"
+        common_utils_version = System.getProperty("common_utils.version", opensearch_build)
         if (buildVersionQualifier) {
             opensearch_build += "-${buildVersionQualifier}"
         }

common/build.gradle

@@ -35,7 +35,8 @@ dependencies {
     api "org.antlr:antlr4-runtime:4.7.1"
     api group: 'com.google.guava', name: 'guava', version: '31.0.1-jre'
     api group: 'org.apache.logging.log4j', name: 'log4j-core', version:'2.17.1'
-    api group: 'org.apache.commons', name: 'commons-lang3', version: '3.10'
+    api group: 'org.apache.commons', name: 'commons-lang3', version: '3.12.0'
+    api 'com.amazonaws:aws-encryption-sdk-java:2.4.0'
 
     testImplementation group: 'junit', name: 'junit', version: '4.13.2'
     testImplementation group: 'org.assertj', name: 'assertj-core', version: '3.9.1'

common/src/main/java/org/opensearch/sql/common/encryptor/Encryptor.java

@@ -0,0 +1,28 @@
+/*
+ *
+ *  * Copyright OpenSearch Contributors
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.opensearch.sql.common.encryptor;
+
+public interface Encryptor {
+
+  /**
+   * Takes plaintext and returns encrypted text.
+   *
+   * @param plainText plainText.
+   * @return String encryptedText.
+   */
+  String encrypt(String plainText);
+
+  /**
+   * Takes encryptedText and returns plain text.
+   *
+   * @param encryptedText encryptedText.
+   * @return String plainText.
+   */
+  String decrypt(String encryptedText);
+
+}

common/src/main/java/org/opensearch/sql/common/encryptor/EncryptorImpl.java

@@ -0,0 +1,55 @@
+/*
+ *
+ *  * Copyright OpenSearch Contributors
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.opensearch.sql.common.encryptor;
+
+import com.amazonaws.encryptionsdk.AwsCrypto;
+import com.amazonaws.encryptionsdk.CommitmentPolicy;
+import com.amazonaws.encryptionsdk.CryptoResult;
+import com.amazonaws.encryptionsdk.jce.JceMasterKey;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import javax.crypto.spec.SecretKeySpec;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class EncryptorImpl implements Encryptor {
+
+  private final String masterKey;
+
+  @Override
+  public String encrypt(String plainText) {
+
+    final AwsCrypto crypto = AwsCrypto.builder()
+        .withCommitmentPolicy(CommitmentPolicy.RequireEncryptRequireDecrypt)
+        .build();
+
+    JceMasterKey jceMasterKey
+        = JceMasterKey.getInstance(new SecretKeySpec(masterKey.getBytes(), "AES"), "Custom", "",
+        "AES/GCM/NoPadding");
+
+    final CryptoResult<byte[], JceMasterKey> encryptResult = crypto.encryptData(jceMasterKey,
+        plainText.getBytes(StandardCharsets.UTF_8));
+    return Base64.getEncoder().encodeToString(encryptResult.getResult());
+  }
+
+  @Override
+  public String decrypt(String encryptedText) {
+    final AwsCrypto crypto = AwsCrypto.builder()
+        .withCommitmentPolicy(CommitmentPolicy.RequireEncryptRequireDecrypt)
+        .build();
+
+    JceMasterKey jceMasterKey
+        = JceMasterKey.getInstance(new SecretKeySpec(masterKey.getBytes(), "AES"), "Custom", "",
+        "AES/GCM/NoPadding");
+
+    final CryptoResult<byte[], JceMasterKey> decryptedResult
+        = crypto.decryptData(jceMasterKey, Base64.getDecoder().decode(encryptedText));
+    return new String(decryptedResult.getResult());
+  }
+
+}

core/build.gradle

@@ -35,12 +35,13 @@ repositories {
 
 dependencies {
     api group: 'com.google.guava', name: 'guava', version: '31.0.1-jre'
-    api group: 'org.apache.commons', name: 'commons-lang3', version: '3.10'
+    api group: 'org.apache.commons', name: 'commons-lang3', version: '3.12.0'
     api group: 'com.facebook.presto', name: 'presto-matching', version: '0.240'
     api group: 'org.apache.commons', name: 'commons-math3', version: '3.6.1'
     api "com.fasterxml.jackson.core:jackson-core:${versions.jackson}"
     api "com.fasterxml.jackson.core:jackson-databind:${versions.jackson_databind}"
     api "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
+    api group: 'com.google.code.gson', name: 'gson', version: '2.8.9'
     api project(':common')
 
     testImplementation('org.junit.jupiter:junit-jupiter:5.6.2')

core/src/main/java/org/opensearch/sql/analysis/Analyzer.java

@@ -25,7 +25,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.commons.lang3.tuple.Pair;
@@ -64,7 +63,6 @@
 import org.opensearch.sql.data.model.ExprMissingValue;
 import org.opensearch.sql.data.type.ExprCoreType;
 import org.opensearch.sql.datasource.DataSourceService;
-import org.opensearch.sql.datasource.model.DataSourceMetadata;
 import org.opensearch.sql.exception.SemanticCheckException;
 import org.opensearch.sql.expression.DSL;
 import org.opensearch.sql.expression.Expression;
@@ -134,13 +132,8 @@ public LogicalPlan analyze(UnresolvedPlan unresolved, AnalysisContext context) {
   @Override
   public LogicalPlan visitRelation(Relation node, AnalysisContext context) {
     QualifiedName qualifiedName = node.getTableQualifiedName();
-    Set<String> allowedDataSourceNames = dataSourceService.getDataSourceMetadataSet()
-        .stream()
-        .map(DataSourceMetadata::getName)
-        .collect(Collectors.toSet());
     DataSourceSchemaIdentifierNameResolver dataSourceSchemaIdentifierNameResolver
-        = new DataSourceSchemaIdentifierNameResolver(qualifiedName.getParts(),
-        allowedDataSourceNames);
+        = new DataSourceSchemaIdentifierNameResolver(dataSourceService, qualifiedName.getParts());
     String tableName = dataSourceSchemaIdentifierNameResolver.getIdentifierName();
     context.push();
     TypeEnvironment curEnv = context.peek();
@@ -182,13 +175,9 @@ public LogicalPlan visitRelationSubquery(RelationSubquery node, AnalysisContext
   @Override
   public LogicalPlan visitTableFunction(TableFunction node, AnalysisContext context) {
     QualifiedName qualifiedName = node.getFunctionName();
-    Set<String> allowedDataSourceNames = dataSourceService.getDataSourceMetadataSet()
-        .stream()
-        .map(DataSourceMetadata::getName)
-        .collect(Collectors.toSet());
     DataSourceSchemaIdentifierNameResolver dataSourceSchemaIdentifierNameResolver
-        = new DataSourceSchemaIdentifierNameResolver(qualifiedName.getParts(),
-        allowedDataSourceNames);
+        = new DataSourceSchemaIdentifierNameResolver(this.dataSourceService,
+        qualifiedName.getParts());
 
     FunctionName functionName
         = FunctionName.of(dataSourceSchemaIdentifierNameResolver.getIdentifierName());

core/src/main/java/org/opensearch/sql/analysis/DataSourceSchemaIdentifierNameResolver.java

@@ -7,8 +7,13 @@
 
 package org.opensearch.sql.analysis;
 
+import com.google.common.collect.ImmutableSet;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
+import org.opensearch.sql.ast.expression.QualifiedName;
+import org.opensearch.sql.datasource.DataSourceService;
+import org.opensearch.sql.datasource.model.DataSourceMetadata;
 
 public class DataSourceSchemaIdentifierNameResolver {
 
@@ -19,6 +24,7 @@ public class DataSourceSchemaIdentifierNameResolver {
   private String dataSourceName = DEFAULT_DATASOURCE_NAME;
   private String schemaName = DEFAULT_SCHEMA_NAME;
   private String identifierName;
+  private DataSourceService dataSourceService;
 
   private static final String DOT = ".";
 
@@ -28,13 +34,14 @@ public class DataSourceSchemaIdentifierNameResolver {
    * DataSourceSchemaTable name and DataSourceSchemaFunction in case of table
    * functions.
    *
+   * @param dataSourceService {@link DataSourceService}.
    * @param parts           parts of qualifiedName.
-   * @param allowedDataSources allowedDataSources.
    */
-  public DataSourceSchemaIdentifierNameResolver(List<String> parts,
-                                                Set<String> allowedDataSources) {
+  public DataSourceSchemaIdentifierNameResolver(DataSourceService dataSourceService,
+                                                List<String> parts) {
+    this.dataSourceService = dataSourceService;
     List<String> remainingParts
-        = captureSchemaName(captureDataSourceName(parts, allowedDataSources));
+        = captureSchemaName(captureDataSourceName(parts));
     identifierName = String.join(DOT, remainingParts);
   }
 
@@ -53,9 +60,8 @@ public String getSchemaName() {
 
   // Capture datasource name and return remaining parts(schema name and table name)
   // from the fully qualified name.
-  private List<String> captureDataSourceName(List<String> parts, Set<String> allowedDataSources) {
-    if (parts.size() > 1 && allowedDataSources.contains(parts.get(0))
-        || DEFAULT_DATASOURCE_NAME.equals(parts.get(0))) {
+  private List<String> captureDataSourceName(List<String> parts) {
+    if (parts.size() > 1 && dataSourceService.dataSourceExists(parts.get(0))) {
       dataSourceName = parts.get(0);
       return parts.subList(1, parts.size());
     } else {
@@ -76,5 +82,4 @@ private List<String> captureSchemaName(List<String> parts) {
     }
   }
 
-
 }

core/src/main/java/org/opensearch/sql/datasource/DataSourceService.java

@@ -34,9 +34,9 @@ public interface DataSourceService {
   /**
    * Register {@link DataSource} defined by {@link DataSourceMetadata}.
    *
-   * @param metadatas list of {@link DataSourceMetadata}.
+   * @param metadata {@link DataSourceMetadata}.
    */
-  void createDataSource(DataSourceMetadata... metadatas);
+  void createDataSource(DataSourceMetadata metadata);
 
   /**
    * Updates {@link DataSource} corresponding to dataSourceMetadata.
@@ -54,13 +54,10 @@ public interface DataSourceService {
   void deleteDataSource(String dataSourceName);
 
   /**
-   * This method is to bootstrap
-   * datasources during the startup of the plugin.
-   */
-  void bootstrapDataSources();
-
-  /**
-   * remove all the registered {@link DataSource}.
+   * Returns true {@link Boolean} if datasource with dataSourceName exists
+   * or else false {@link Boolean}.
+   *
+   * @param dataSourceName name of the {@link DataSource}.
    */
-  void clear();
+  Boolean dataSourceExists(String dataSourceName);
 }