New Interface for DataSource Management

common/build.gradle

@@ -35,7 +35,7 @@ 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'
 
     testImplementation group: 'junit', name: 'junit', version: '4.13.2'
     testImplementation group: 'org.assertj', name: 'assertj-core', version: '3.9.1'

core/build.gradle

@@ -37,12 +37,14 @@ dependencies {
     api group: 'com.google.guava', name: 'guava', version: '31.0.1-jre'
     api group: 'org.springframework', name: 'spring-context', version: "${spring_version}"
     api group: 'org.springframework', name: 'spring-beans', version: "${spring_version}"
-    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'
+    implementation 'com.amazonaws:aws-encryption-sdk-java:2.4.0'
     api project(':common')
 
     testImplementation('org.junit.jupiter:junit-jupiter:5.6.2')

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

@@ -64,7 +64,7 @@
 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.DataSource;
+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,9 +134,9 @@ public LogicalPlan analyze(UnresolvedPlan unresolved, AnalysisContext context) {
   @Override
   public LogicalPlan visitRelation(Relation node, AnalysisContext context) {
     QualifiedName qualifiedName = node.getTableQualifiedName();
-    Set<String> allowedDataSourceNames = dataSourceService.getDataSources()
+    Set<String> allowedDataSourceNames = dataSourceService.getMaskedDataSourceMetadataInfo()
         .stream()
-        .map(DataSource::getName)
+        .map(DataSourceMetadata::getName)
         .collect(Collectors.toSet());
     DataSourceSchemaIdentifierNameResolver dataSourceSchemaIdentifierNameResolver
         = new DataSourceSchemaIdentifierNameResolver(qualifiedName.getParts(),
@@ -182,9 +182,9 @@ public LogicalPlan visitRelationSubquery(RelationSubquery node, AnalysisContext
   @Override
   public LogicalPlan visitTableFunction(TableFunction node, AnalysisContext context) {
     QualifiedName qualifiedName = node.getFunctionName();
-    Set<String> allowedDataSourceNames = dataSourceService.getDataSources()
+    Set<String> allowedDataSourceNames = dataSourceService.getMaskedDataSourceMetadataInfo()
         .stream()
-        .map(DataSource::getName)
+        .map(DataSourceMetadata::getName)
         .collect(Collectors.toSet());
     DataSourceSchemaIdentifierNameResolver dataSourceSchemaIdentifierNameResolver
         = new DataSourceSchemaIdentifierNameResolver(qualifiedName.getParts(),

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

@@ -0,0 +1,25 @@
+/*
+ *
+ *  * Copyright OpenSearch Contributors
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.opensearch.sql.datasource;
+
+import java.util.List;
+import java.util.Optional;
+import org.opensearch.sql.datasource.model.DataSourceMetadata;
+
+/**
+ * Interface for DataSourceMetadata Storage.
+ */
+public interface DataSourceMetadataStorage {
+
+  List<DataSourceMetadata> getDataSourceMetadata();
+
+  Optional<DataSourceMetadata> getDataSourceMetadata(String datasourceName);
+
+  void createDataSourceMetadata(DataSourceMetadata dataSourceMetadata);
+
+}

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

@@ -19,7 +19,7 @@ public interface DataSourceService {
    *
    * @return set of {@link DataSource}.
    */
-  Set<DataSource> getDataSources();
+  Set<DataSourceMetadata> getMaskedDataSourceMetadataInfo();
 
   /**
    * Returns {@link DataSource} with corresponding to the DataSource name.
@@ -36,6 +36,7 @@ public interface DataSourceService {
    */
   void addDataSource(DataSourceMetadata... metadatas);
 
+
   /**
    * remove all the registered {@link DataSource}.
    */

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

@@ -5,14 +5,16 @@
 
 package org.opensearch.sql.datasource;
 
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
+import static org.opensearch.sql.analysis.DataSourceSchemaIdentifierNameResolver.DEFAULT_DATASOURCE_NAME;
+
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.HashSet;
 import java.util.Map;
-import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
-import org.opensearch.sql.common.utils.StringUtils;
 import org.opensearch.sql.datasource.model.DataSource;
 import org.opensearch.sql.datasource.model.DataSourceMetadata;
 import org.opensearch.sql.datasource.model.DataSourceType;
@@ -28,72 +30,68 @@
  */
 public class DataSourceServiceImpl implements DataSourceService {
 
-  private static String DATASOURCE_NAME_REGEX = "[@*A-Za-z]+?[*a-zA-Z_\\-0-9]*";
-
-  private final ConcurrentHashMap<String, DataSource> dataSourceMap;
+  private final ConcurrentHashMap<DataSourceMetadata, DataSource> dataSourceMap;
 
   private final Map<DataSourceType, DataSourceFactory> dataSourceFactoryMap;
 
+  private final DataSourceMetadataStorage dataSourceMetadataStorage;
+
+  private final String clusterName;
+
   /**
    * Construct from the set of {@link DataSourceFactory} at bootstrap time.
    */
-  public DataSourceServiceImpl(Set<DataSourceFactory> dataSourceFactories) {
+  public DataSourceServiceImpl(DataSourceMetadataStorage dataSourceMetadataStorage,
+                               Set<DataSourceFactory> dataSourceFactories,
+                               String clusterName) {
+    this.dataSourceMetadataStorage = dataSourceMetadataStorage;
     dataSourceFactoryMap =
         dataSourceFactories.stream()
             .collect(Collectors.toMap(DataSourceFactory::getDataSourceType, f -> f));
     dataSourceMap = new ConcurrentHashMap<>();
+    this.clusterName = clusterName;
+
   }
 
   @Override
-  public Set<DataSource> getDataSources() {
-    return Set.copyOf(dataSourceMap.values());
+  public Set<DataSourceMetadata> getMaskedDataSourceMetadataInfo() {
+    return new HashSet<>(this.dataSourceMetadataStorage.getDataSourceMetadata());
   }
 
   @Override
   public DataSource getDataSource(String dataSourceName) {
-    if (!dataSourceMap.containsKey(dataSourceName)) {
-      throw new IllegalArgumentException(
-          String.format("DataSource with name %s doesn't exist.", dataSourceName));
+    Optional<DataSourceMetadata> dataSourceMetadataOptional
+        = this.dataSourceMetadataStorage.getDataSourceMetadata(dataSourceName);
+    if (dataSourceMetadataOptional.isEmpty()) {
+      if (dataSourceName.equals(DEFAULT_DATASOURCE_NAME)) {
+        return dataSourceMap.get(DataSourceMetadata.defaultOpenSearchDataSourceMetadata());
+      } else {
+        throw new IllegalArgumentException(
+            String.format("DataSource with name %s doesn't exist.", dataSourceName));
+      }
+    } else if (!dataSourceMap.containsKey(dataSourceMetadataOptional.get())) {
+      clearDataSource(dataSourceMetadataOptional.get());
+      addDataSource(dataSourceMetadataOptional.get());
     }
-    return dataSourceMap.get(dataSourceName);
+    return dataSourceMap.get(dataSourceMetadataOptional.get());
+  }
+
+  private void clearDataSource(DataSourceMetadata dataSourceMetadata) {
+    dataSourceMap.entrySet()
+        .removeIf(entry -> entry.getKey().getName().equals(dataSourceMetadata.getName()));
   }
 
   @Override
   public void addDataSource(DataSourceMetadata... metadatas) {
     for (DataSourceMetadata metadata : metadatas) {
-      validateDataSourceMetaData(metadata);
-      dataSourceMap.put(
-          metadata.getName(),
-          dataSourceFactoryMap.get(metadata.getConnector()).createDataSource(metadata));
+      AccessController.doPrivileged((PrivilegedAction<DataSource>) () -> dataSourceMap.put(
+          metadata,
+          dataSourceFactoryMap.get(metadata.getConnector()).createDataSource(metadata, clusterName)));
     }
   }
 
   @Override
   public void clear() {
     dataSourceMap.clear();
   }
-
-  /**
-   * This can be moved to a different validator class when we introduce more connectors.
-   *
-   * @param metadata {@link DataSourceMetadata}.
-   */
-  private void validateDataSourceMetaData(DataSourceMetadata metadata) {
-    Preconditions.checkArgument(
-        !Strings.isNullOrEmpty(metadata.getName()),
-        "Missing Name Field from a DataSource. Name is a required parameter.");
-    Preconditions.checkArgument(
-        !dataSourceMap.containsKey(metadata.getName()),
-        StringUtils.format(
-            "Datasource name should be unique, Duplicate datasource found %s.",
-            metadata.getName()));
-    Preconditions.checkArgument(
-        metadata.getName().matches(DATASOURCE_NAME_REGEX),
-        StringUtils.format(
-            "DataSource Name: %s contains illegal characters. Allowed characters: a-zA-Z0-9_-*@.",
-            metadata.getName()));
-    Preconditions.checkArgument(
-        !Objects.isNull(metadata.getProperties()),
-        "Missing properties field in catalog configuration. Properties are required parameters.");
-  }
 }

core/src/main/java/org/opensearch/sql/datasource/encryptor/CredentialInfoEncryptor.java

@@ -0,0 +1,16 @@
+/*
+ *
+ *  * Copyright OpenSearch Contributors
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.opensearch.sql.datasource.encryptor;
+
+public interface CredentialInfoEncryptor {
+
+  String encrypt(String plainText);
+
+  String decrypt(String encryptedText);
+
+}

core/src/main/java/org/opensearch/sql/datasource/encryptor/CredentialInfoEncryptorImpl.java

@@ -0,0 +1,55 @@
+/*
+ *
+ *  * Copyright OpenSearch Contributors
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.opensearch.sql.datasource.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 CredentialInfoEncryptorImpl implements CredentialInfoEncryptor {
+
+  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/src/main/java/org/opensearch/sql/planner/physical/datasource/DataSourceTableScan.java

@@ -18,7 +18,7 @@
 import org.opensearch.sql.data.model.ExprValue;
 import org.opensearch.sql.data.model.ExprValueUtils;
 import org.opensearch.sql.datasource.DataSourceService;
-import org.opensearch.sql.datasource.model.DataSource;
+import org.opensearch.sql.datasource.model.DataSourceMetadata;
 import org.opensearch.sql.storage.TableScanOperator;
 
 /**
@@ -47,14 +47,15 @@ public String explain() {
   @Override
   public void open() {
     List<ExprValue> exprValues = new ArrayList<>();
-    Set<DataSource> dataSources = dataSourceService.getDataSources();
-    for (DataSource dataSource : dataSources) {
+    Set<DataSourceMetadata> dataSourceMetadataSet
+        = dataSourceService.getMaskedDataSourceMetadataInfo();
+    for (DataSourceMetadata dataSourceMetadata : dataSourceMetadataSet) {
       exprValues.add(
           new ExprTupleValue(new LinkedHashMap<>(ImmutableMap.of(
               "DATASOURCE_NAME",
-              ExprValueUtils.stringValue(dataSource.getName()),
+              ExprValueUtils.stringValue(dataSourceMetadata.getName()),
               "CONNECTOR_TYPE",
-              ExprValueUtils.stringValue(dataSource.getConnectorType().name())))));
+              ExprValueUtils.stringValue(dataSourceMetadata.getConnector().name())))));
     }
     iterator = exprValues.iterator();
   }

core/src/main/java/org/opensearch/sql/storage/DataSourceFactory.java

@@ -27,5 +27,5 @@ public interface DataSourceFactory {
   /**
    * Create {@link DataSource}.
    */
-  DataSource createDataSource(DataSourceMetadata metadata);
+  DataSource createDataSource(DataSourceMetadata metadata, String clusterName);
 }

integ-test/build.gradle

@@ -75,6 +75,8 @@ configurations.all {
     resolutionStrategy.force "org.apache.httpcomponents:httpcore:4.4.13"
     resolutionStrategy.force "joda-time:joda-time:2.10.12"
     resolutionStrategy.force "org.slf4j:slf4j-api:1.7.36"
+    resolutionStrategy.force "org.apache.httpcomponents:httpcore:4.4.15"
+    resolutionStrategy.force "org.apache.httpcomponents:httpclient:4.5.13"
 }
 
 dependencies {

legacy/build.gradle

@@ -90,7 +90,7 @@ dependencies {
     }
     implementation group: 'com.google.guava', name: 'guava', version: '31.0.1-jre'
     implementation group: 'org.json', name: 'json', version:'20180813'
-    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.10'
+    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.12.0'
     implementation group: 'org.opensearch', name: 'opensearch', version: "${opensearch_version}"
     // add geo module as dependency. https://github.com/opensearch-project/OpenSearch/pull/4180/.
     implementation group: 'org.opensearch.plugin', name: 'geo', version: "${opensearch_version}"

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchDataSourceFactory.java

@@ -27,7 +27,7 @@ public DataSourceType getDataSourceType() {
   }
 
   @Override
-  public DataSource createDataSource(DataSourceMetadata metadata) {
+  public DataSource createDataSource(DataSourceMetadata metadata, String clusterName) {
     return new DataSource(metadata.getName(), DataSourceType.OPENSEARCH,
         new OpenSearchStorageEngine(client, settings));
   }

plugin/build.gradle

@@ -98,6 +98,8 @@ configurations.all {
     resolutionStrategy.force "com.squareup.okhttp3:okhttp:4.9.3"
     resolutionStrategy.force "joda-time:joda-time:2.10.12"
     resolutionStrategy.force "org.slf4j:slf4j-api:1.7.36"
+    resolutionStrategy.force "org.apache.httpcomponents:httpcore:4.4.15"
+    resolutionStrategy.force "org.apache.httpcomponents:httpclient:4.5.13"
 }
 compileJava {
     options.compilerArgs.addAll(["-processor", 'lombok.launch.AnnotationProcessorHider$AnnotationProcessor'])
@@ -112,6 +114,7 @@ dependencies {
     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}"
+    implementation group: 'commons-io', name: 'commons-io', version: '2.8.0'
 
     api project(":ppl")
     api project(':legacy')