Merge branch 'gcp-stabilize-idexer-reference-with-postman-collection' into 'master'

Stabilize Indexer Service With Postman Collection (GONRG-3817)

See merge request !236

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/cache/DatastoreCredentialCache.java

-// Copyright 2017-2019, Schlumberger
-//
-// Licensed 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.
+/*
+ * Copyright 2021 Google LLC
+ * Copyright 2021 EPAM Systems, Inc
+ *
+ * Licensed 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
+ *
+ *     https://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.opengroup.osdu.indexer.cache;
 

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/cache/JwtCache.java

@@ -29,7 +29,7 @@ public class JwtCache implements IJwtCache<String, IdToken>, AutoCloseable {
   RedisCache<String, IdToken> cache;
 
   // google service account id_token can be requested only for 1 hr
-  private final static int EXPIRED_AFTER = 59;
+  private static final int EXPIRED_AFTER = 59;
 
   public JwtCache(IndexerConfigurationProperties indexerConfigurationProperties) {
     cache = new RedisCache<>(indexerConfigurationProperties.getRedisSearchHost(),

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/config/EntitlementsConfigProperties.java

@@ -28,5 +28,5 @@ import org.springframework.context.annotation.Configuration;
 @Setter
 public class EntitlementsConfigProperties {
 
-  private String AuthorizeApi;
+  private String authorizeApi;
 }

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/config/IndexerReferenceConfigProperties.java

+/*
+ * Copyright 2021 Google LLC
+ * Copyright 2021 EPAM Systems, Inc
+ *
+ * Licensed 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
+ *
+ *     https://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.opengroup.osdu.indexer.config;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@ConfigurationProperties
+@Getter
+@Setter
+public class IndexerReferenceConfigProperties {
+
+  private String elasticSettingsId;
+
+}

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/di/TenantInfoService.java

+/*
+ * Copyright 2021 Google LLC
+ * Copyright 2021 EPAM Systems, Inc
+ *
+ * Licensed 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
+ *
+ *     https://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.opengroup.osdu.indexer.di;
 
 import java.util.ArrayList;

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/model/ElasticSettingSchema.java

+/*
+ * Copyright 2021 Google LLC
+ * Copyright 2021 EPAM Systems, Inc
+ *
+ * Licensed 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
+ *
+ *     https://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.opengroup.osdu.indexer.model;
+
+import javax.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class ElasticSettingSchema {
+
+  @NotEmpty
+  private String host;
+
+  @NotEmpty
+  private String port;
+
+  @NotEmpty
+  private String usernameAndPassword;
+
+  @NotEmpty
+  private boolean isHttps;
+
+}

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/model/ElasticSettingsDoc.java

+/*
+ * Copyright 2021 Google LLC
+ * Copyright 2021 EPAM Systems, Inc
+ *
+ * Licensed 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
+ *
+ *     https://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.opengroup.osdu.indexer.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class ElasticSettingsDoc {
+
+  @Id
+  private String id;
+  private ElasticSettingSchema settingSchema;
+
+  public void setId(String id) {
+    this.id = id;
+  }
+
+}

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/persistence/ElasticRepositoryMongoDB.java

@@ -17,71 +17,89 @@
 
 package org.opengroup.osdu.indexer.persistence;
 
-import com.mongodb.client.FindIterable;
+import static com.mongodb.client.model.Filters.eq;
+import static com.mongodb.util.JSON.serialize;
+
+import com.google.gson.Gson;
 import com.mongodb.client.MongoCollection;
 import java.util.Objects;
+import lombok.RequiredArgsConstructor;
 import org.apache.http.HttpStatus;
 import org.bson.Document;
 import org.opengroup.osdu.core.common.model.http.AppException;
 import org.opengroup.osdu.core.common.model.search.ClusterSettings;
 import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
 import org.opengroup.osdu.core.common.provider.interfaces.IElasticRepository;
+import org.opengroup.osdu.core.common.search.Preconditions;
+import org.opengroup.osdu.indexer.config.IndexerReferenceConfigProperties;
+import org.opengroup.osdu.indexer.model.ElasticSettingsDoc;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@RequiredArgsConstructor
 public class ElasticRepositoryMongoDB implements IElasticRepository {
 
   private static final Logger LOG = LoggerFactory.getLogger(ElasticRepositoryMongoDB.class);
 
-  private static final String SCHEMA_DATABASE = "local";
-  private static final String SEARCH_SETTINGS = "SearchSettings";
-  private static final String HOST = "host";
-  private static final String PORT = "port";
-  private static final String XPACK_RESTCLIENT_CONFIGURATION = "configuration";
+  private static final String ELASTIC_SETTINGS_DATABASE = "local";
+  private static final String ELASTIC_SETTINGS_COLLECTION = "SearchSettings";
 
-  private final MongoDdmsClient mongoClient;
+  private static final String MISSING_TENANT_INFO_REASON = "TenantInfo is null";
+  private static final String MISSING_TENANT_INFO_MESSAGE = "TenantInfo is missing.";
+  private static final String SETTINGS_FETCH_ERROR_REASON = "Cluster setting fetch error";
+  private static final String SETTINGS_FETCH_ERROR_MESSAGE = "An error has occurred fetching cluster settings from the database.";
 
-  @Autowired
-  public ElasticRepositoryMongoDB(MongoDdmsClient mongoClient) {
-    this.mongoClient = mongoClient;
-  }
+  private final MongoDdmsClient mongoClient;
+  private final IndexerReferenceConfigProperties configProperties;
 
   @Override
   public ClusterSettings getElasticClusterSettings(TenantInfo tenantInfo) {
-    MongoCollection<Document> mongoCollection = mongoClient
-        .getMongoCollection(SCHEMA_DATABASE, SEARCH_SETTINGS);
-
-    FindIterable<Document> results = mongoCollection.find();
 
-    if (Objects.isNull(results) && Objects.isNull(results.first())) {
-      LOG.error(String.format("Collection \'%s\' is empty.", results));
-      throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Cluster setting fetch error",
-          "An error has occurred fetching cluster settings from the database.");
+    if (Objects.isNull(tenantInfo)) {
+      throw new AppException(HttpStatus.SC_NOT_FOUND, MISSING_TENANT_INFO_REASON,
+          MISSING_TENANT_INFO_MESSAGE);
     }
 
-    Document document = results.first();
+    String settingId = tenantInfo.getName().concat("-")
+        .concat(this.configProperties.getElasticSettingsId());
 
-    String encryptedConfiguration = document.get(XPACK_RESTCLIENT_CONFIGURATION).toString();
-    String encryptedHost = document.get(HOST).toString();
-    String encryptedPort = document.get(PORT).toString();
+    MongoCollection<Document> collection = this.mongoClient.getMongoCollection(
+        ELASTIC_SETTINGS_DATABASE, ELASTIC_SETTINGS_COLLECTION);
+    Document record = collection.find(eq("_id", settingId)).first();
+
+    if (Objects.isNull(record)) {
+      LOG.error(String.format("\'%s\' credentials not found in database.", settingId));
+      throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, SETTINGS_FETCH_ERROR_REASON,
+          SETTINGS_FETCH_ERROR_MESSAGE);
+    }
+
+    ElasticSettingsDoc elasticSettingsDoc = new Gson().fromJson(serialize(record),
+        ElasticSettingsDoc.class);
 
     try {
+      String encryptedUsernameAndPassword = elasticSettingsDoc.getSettingSchema()
+          .getUsernameAndPassword();
+      String encryptedHost = elasticSettingsDoc.getSettingSchema().getHost();
+      String encryptedPortString = elasticSettingsDoc.getSettingSchema().getPort();
 
-      String host = encryptedHost;//this.kmsClient.decryptString(encryptedHost);
-      String portString = encryptedPort;//this.kmsClient.decryptString(encryptedPort);
-      String usernameAndPassword = encryptedConfiguration;//this.kmsClient.decryptString(encryptedConfiguration);
+      String usernameAndPassword = encryptedUsernameAndPassword; //this.kmsClient.decryptString(encryptedHost);
+      String host = encryptedHost; //this.kmsClient.decryptString(encryptedHost);
+      String portString = encryptedPortString; //this.kmsClient.decryptString(encryptedHost);
+
+      Preconditions.checkNotNullOrEmpty(usernameAndPassword, "configuration cannot be null");
+      Preconditions.checkNotNullOrEmpty(host, "host cannot be null");
+      Preconditions.checkNotNullOrEmpty(portString, "port cannot be null");
 
       int port = Integer.parseInt(portString);
-      ClusterSettings clusterSettings = new ClusterSettings(host, port, usernameAndPassword);
-      clusterSettings.setHttps(false);
-      return clusterSettings;
+      boolean isHttps = elasticSettingsDoc.getSettingSchema().isHttps();
 
+      return new ClusterSettings(host, port, usernameAndPassword, isHttps, isHttps);
     } catch (Exception e) {
-      throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Cluster setting fetch error",
-          "An error has occurred fetching cluster settings from the database.", e);
+      throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, SETTINGS_FETCH_ERROR_REASON,
+          SETTINGS_FETCH_ERROR_MESSAGE, e);
     }
   }
+
 }

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/persistence/MongoDdmsClient.java

@@ -19,7 +19,6 @@ package org.opengroup.osdu.indexer.persistence;
 
 import com.mongodb.client.MongoCollection;
 import org.bson.Document;
-import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
 import org.opengroup.osdu.indexer.util.MongoClientHandler;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/service/IndexerServiceImpl.java

@@ -17,6 +17,7 @@
 
 package org.opengroup.osdu.indexer.service;
 
+import com.google.common.base.Strings;
 import com.google.gson.Gson;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -75,6 +76,8 @@ public class IndexerServiceImpl implements IndexerService {
       .asList(RestStatus.TOO_MANY_REQUESTS, RestStatus.BAD_GATEWAY,
           RestStatus.SERVICE_UNAVAILABLE));
 
+  public static final String ELASTIC_ERROR_REASON = "Elastic error";
+
   private final Gson gson = new Gson();
 
   @Inject
@@ -111,7 +114,7 @@ public class IndexerServiceImpl implements IndexerService {
       List<RecordInfo> recordInfos) throws Exception {
 
     // this should not happen
-    if (recordInfos.size() == 0) {
+    if (recordInfos.isEmpty()) {
       return null;
     }
 
@@ -121,7 +124,7 @@ public class IndexerServiceImpl implements IndexerService {
     // get auth header with service account Authorization
     this.headers = this.requestInfo.getHeaders();
     String authorization = message.getAttributes().get("authorization");
-    headers.put("authorization", authorization);
+    this.headers.put("authorization", authorization);
     // initialize status for all messages.
     this.jobStatus.initialize(recordInfos);
 
@@ -152,7 +155,7 @@ public class IndexerServiceImpl implements IndexerService {
       }
 
       // process failed records
-      if (retryRecordIds.size() > 0) {
+      if (!retryRecordIds.isEmpty()) {
         retryAndEnqueueFailedRecords(recordInfos, retryRecordIds, message);
       }
     } catch (IOException e) {
@@ -375,6 +378,14 @@ public class IndexerServiceImpl implements IndexerService {
       if (recordStatus.getIndexProgress().getStatusCode() == 0) {
         recordStatus.getIndexProgress().setStatusCode(HttpStatus.SC_OK);
       }
+      document.setCreateUser(storageRecord.getCreateUser());
+      document.setCreateTime(storageRecord.getCreateTime());
+      if (!Strings.isNullOrEmpty(storageRecord.getModifyUser())) {
+        document.setModifyUser(storageRecord.getModifyUser());
+      }
+      if (!Strings.isNullOrEmpty(storageRecord.getModifyTime())) {
+        document.setModifyTime(storageRecord.getModifyTime());
+      }
       document.setIndexProgress(recordStatus.getIndexProgress());
       if (storageRecord.getAncestry() != null) {
         document.setAncestry(storageRecord.getAncestry());
@@ -413,15 +424,16 @@ public class IndexerServiceImpl implements IndexerService {
     for (IndexSchema schema : schemas) {
       String index = this.elasticIndexNameResolver.getIndexNameFromKind(schema.getKind());
 
-      // check if index exist
+      // check if index exist and sync meta attribute schema if required
       if (this.indicesService.isIndexExist(restClient, index)) {
+        this.mappingService.syncIndexMappingIfRequired(restClient, index);
         continue;
       }
 
       // create index
       Map<String, Object> mapping = this.mappingService.getIndexMappingFromRecordSchema(schema);
       if (!this.indicesService.createIndex(restClient, index, null, schema.getType(), mapping)) {
-        throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Elastic error",
+        throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, ELASTIC_ERROR_REASON,
             "Error creating index.",
             String.format("Failed to get confirmation from elastic server for index: %s", index));
       }
@@ -469,10 +481,6 @@ public class IndexerServiceImpl implements IndexerService {
     bulkRequest.timeout(BULK_REQUEST_TIMEOUT);
 
     for (Map.Entry<String, List<String>> record : deleteRecordMap.entrySet()) {
-
-      String[] kindParts = record.getKey().split(":");
-      String type = kindParts[2];
-
       String index = this.elasticIndexNameResolver.getIndexNameFromKind(record.getKey());
 
       for (String id : record.getValue()) {
@@ -528,12 +536,12 @@ public class IndexerServiceImpl implements IndexerService {
               bulkRequest.numberOfActions(), succeededResponses, failedResponses));
     } catch (IOException e) {
       // throw explicit 504 for IOException
-      throw new AppException(HttpStatus.SC_GATEWAY_TIMEOUT, "Elastic error",
+      throw new AppException(HttpStatus.SC_GATEWAY_TIMEOUT, ELASTIC_ERROR_REASON,
           "Request cannot be completed in specified time.", e);
     } catch (ElasticsearchStatusException e) {
-      throw new AppException(e.status().getStatus(), "Elastic error", e.getMessage(), e);
+      throw new AppException(e.status().getStatus(), ELASTIC_ERROR_REASON, e.getMessage(), e);
     } catch (Exception e) {
-      throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Elastic error",
+      throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, ELASTIC_ERROR_REASON,
           "Error indexing records.", e);
     }
     return failureRecordIds;
@@ -554,6 +562,8 @@ public class IndexerServiceImpl implements IndexerService {
 
     indexerPayload.put(RecordMetaAttribute.ID.getValue(), record.getId());
     indexerPayload.put(RecordMetaAttribute.KIND.getValue(), record.getKind());
+    indexerPayload.put(RecordMetaAttribute.AUTHORITY.getValue(), record.getAuthority());
+    indexerPayload.put(RecordMetaAttribute.SOURCE.getValue(), record.getSource());
     indexerPayload.put(RecordMetaAttribute.NAMESPACE.getValue(), record.getNamespace());
     indexerPayload.put(RecordMetaAttribute.TYPE.getValue(), record.getType());
     indexerPayload.put(RecordMetaAttribute.VERSION.getValue(), record.getVersion());
@@ -565,6 +575,14 @@ public class IndexerServiceImpl implements IndexerService {
     if (record.getAncestry() != null) {
       indexerPayload.put(RecordMetaAttribute.ANCESTRY.getValue(), record.getAncestry());
     }
+    indexerPayload.put(RecordMetaAttribute.CREATE_USER.getValue(), record.getCreateUser());
+    indexerPayload.put(RecordMetaAttribute.CREATE_TIME.getValue(), record.getCreateTime());
+    if (!Strings.isNullOrEmpty(record.getModifyUser())) {
+      indexerPayload.put(RecordMetaAttribute.MODIFY_USER.getValue(), record.getModifyUser());
+    }
+    if (!Strings.isNullOrEmpty(record.getModifyTime())) {
+      indexerPayload.put(RecordMetaAttribute.MODIFY_TIME.getValue(), record.getModifyTime());
+    }
     return indexerPayload;
   }
 

provider/indexer-reference/src/main/java/org/opengroup/osdu/indexer/util/ServiceAccountJwtGcpClientImpl.java

@@ -44,11 +44,9 @@ import org.springframework.web.context.annotation.RequestScope;
 @RequestScope
 public class ServiceAccountJwtGcpClientImpl implements IServiceAccountJwtClient {
 
-  private static final String JWT_AUDIENCE = "https://www.googleapis.com/oauth2/v4/token";
   private static final String SERVICE_ACCOUNT_NAME_FORMAT = "projects/%s/serviceAccounts/%s";
   private final IamCredentialsProvider iamCredentialsProvider = new IamCredentialsProvider();
 
-  private final JsonFactory JSON_FACTORY = new JacksonFactory();
   private final IndexerConfigurationProperties indexerConfigurationProperties;
   private final ITenantFactory tenantInfoServiceProvider;
   private final IJwtCache cacheService;

provider/indexer-reference/src/main/resources/application.properties

@@ -43,8 +43,7 @@ gae-service=indexer
 key-ring=csqp
 kms-key=searchservice
 
-elastic-datastore-kind=searchsettings
-elastic-datastore-id=indexer-service
+elastic-settings-id=indexer-service
 
 mongo-db-url=localhost:27017
 mongo-db-user=