Merging all changes from int to master

.github/workflows/reusable-build-and-publish.yml

@@ -45,7 +45,7 @@ jobs:
       ### TODO : Publish to maven central repository
       - name: Prepare CodeDeploy artifact content
         run: |
-          cp -p $SERVICE_JAR_DIR/xyz-hub-service.jar $CODEDEPLOY_DIR/contents/
+          cp -p $SERVICE_JAR_DIR/xyz-hub-service.jar $CODEDEPLOY_DIR/contents/xyz-hub/
       - name: List CodeDeploy artifact content
         run: |
           ls -lR $CODEDEPLOY_DIR

CONTRIBUTING.md

@@ -43,4 +43,4 @@ Eventually, before making a pull request, ensure following:
 
 1. If any **new secret** parameter is introduced, then mention in pull-request as an indication for the repo-owner to add the secret in respective cloud environments.
 2. If any **new config** (not a secret) parameter is introduced in default [config.json](xyz-hub-service/src/main/resources/config.json),
-ensure the same is added into **deployment** [config.json](deployment/codedeploy/contents/.config/config.json) as well.
+ensure the same is added into **deployment** [config.json](deployment/codedeploy/contents/xyz-hub/.config/config.json) as well.

deployment/codedeploy/appspec.yml

@@ -25,8 +25,8 @@ os: linux
 file_exists_behavior: OVERWRITE
 files:
   - source: /contents
-    destination: /home/admin/xyz-hub
-  - source: /contents/xyz-hub.service
+    destination: /home/admin
+  - source: /contents/xyz-hub/xyz-hub.service
     destination: /etc/systemd/system
 # For deployments to Amazon Linux, Ubuntu Server, or RHEL instances,
 #   you can specify a "permissions"
@@ -39,28 +39,38 @@ files:
 #   "permissions" section altogether. A blank or incomplete "permissions"
 #   section may cause associated deployments to fail.
 permissions:
+  # change owner for xyz-hub directory
+  - object: /home/admin
+    pattern: "xyz-hub"
+    owner: admin
+    group: admin
+    type:
+      - directory
+  # change owner for all subdirectories inside xyz-hub
+  - object: /home/admin/xyz-hub
+    owner: admin
+    group: admin
+    type:
+      - directory
+  # change owner for all files in xyz-hub directory (except shell scripts)
   - object: /home/admin/xyz-hub
     pattern: "**"
     except: ["set-app-params.sh","set-auth-keys.sh","set-instance-params.sh","start-app.sh"]
     owner: admin
     group: admin
-    #mode:
-    #acls:
-      #-
-    #context:
-      #user:
-      #type:
-      #range:
     type:
-      - directory
+      - file
+  # change owner+permission for all shell scripts in xyz-hub directory
   - object: /home/admin/xyz-hub
     pattern: "*.sh"
     owner: admin
     group: admin
     mode: 755
     type:
       - file
-  - object: /etc/systemd/system/xyz-hub.service
+  # change permission for file xyz-hub.service
+  - object: /etc/systemd/system
+    pattern: "xyz-hub.service"
     owner: admin
     group: admin
     type:

deployment/codedeploy/contents/.config/config.json → deployment/codedeploy/contents/xyz-hub/.config/config.json

@@ -60,8 +60,8 @@
     "TXN_PUB_TPOOL_MAX_SIZE" : 100,
     "TXN_PUB_TPOOL_KEEP_ALIVE_SEC" : 60,
 
-    "AWS_ACCESS_KEY_ID": "<<to-be-set-as-env-variable>>",
-    "AWS_SECRET_ACCESS_KEY": "<<to-be-set-as-env-variable>>",
+    "AWS_ACCESS_KEY_ID": "",
+    "AWS_SECRET_ACCESS_KEY": "",
     "AWS_DEFAULT_REGION": "<<to-be-set-as-env-variable>>",
 
     "ENABLE_TXN_SEQUENCER" : true,

deployment/codedeploy/scripts/ApplicationStart.sh

@@ -1,4 +1,6 @@
 #!/bin/bash
 
 # Start application
+echo "Starting Naksha service..."
 sudo systemctl start xyz-hub
+echo "Start triggered!"

deployment/codedeploy/scripts/ApplicationStop.sh

@@ -1,4 +1,11 @@
 #!/bin/bash
 
-# Stop application
-sudo systemctl stop xyz-hub
+# Stop Naksha service (only if it was configured)
+CODE=0
+sudo systemctl list-unit-files --all | grep -Fq "xyz-hub.service" || CODE=1
+if [[ $CODE -eq 0 ]]; then
+  echo ">> Stopping service [xyz-hub]..."
+  sudo systemctl stop xyz-hub
+else
+  echo ">> Service [xyz-hub] not found. Nothing to stop"
+fi

test-scripts/xyz-api-maptask-suite.jmx

@@ -836,7 +836,6 @@ var createdTaskStr = vars.get("V_Created_Task_Str");
 var updateTaskObj = JSON.parse(createdTaskStr);
 updateTaskObj.properties["@ns:com:here:xyz"].tags.push("${UTM_DUMMY_UPDATE_TAG}");
 
-
 vars.put("V_Update_Request_Str", JSON.stringify(updateTaskObj));
 </stringProp>
           <stringProp name="scriptLanguage">javascript</stringProp>
@@ -1423,8 +1422,8 @@ vars.put(&quot;V_Update_Request_Str_02&quot;, JSON.stringify(updateTaskObj));
                 <stringProp name="Argument.value">{&#xd;
   &quot;type&quot;: &quot;FeatureCollection&quot;,&#xd;
   &quot;features&quot;: [&#xd;
-    ${V_Update_Request_Str_01},&#xd;
-    ${V_Update_Request_Str_02}&#xd;
+    ${V_Update_Request_Str_02},&#xd;
+    ${V_Update_Request_Str_01}&#xd;
   ]&#xd;
 }&#xd;
 </stringProp>
@@ -1469,10 +1468,10 @@ prev.setSuccessful(false);
 var expectedResObj = {
   &quot;type&quot;: &quot;FeatureCollection&quot;,
   &quot;features&quot;: [
-    ${V_Update_Request_Str_01},
-    ${V_Update_Request_Str_02}
+    ${V_Update_Request_Str_02},
+    ${V_Update_Request_Str_01}
   ],
-  &quot;updated&quot;: [&quot;${TASK_ID_P02_01}&quot;, &quot;${TASK_ID_P02_02}&quot;]
+  &quot;updated&quot;: [&quot;${TASK_ID_P02_02}&quot;, &quot;${TASK_ID_P02_01}&quot;]
 };
 
 // Remove unwanted fields from RESPONSE object, before actual v/s expected comparison

xyz-hub-service/pom.xml

@@ -47,6 +47,7 @@
     <pathToConfig>deployment/local/config.json</pathToConfig>
     <dockerComposeFile>docker-compose.yml</dockerComposeFile>
     <aws-sdk.version>2.20.11</aws-sdk.version>
+    <otel.version>1.28.0</otel.version>
   </properties>
 
   <build>
@@ -326,6 +327,13 @@
       </exclusions>
     </dependency>
 
+    <!-- io.opentelemetry -->
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-api</artifactId>
+      <version>${otel.version}</version>
+    </dependency>
+
     <!-- Cache -->
     <dependency>
       <artifactId>expiringmap</artifactId>

xyz-hub-service/src/main/java/com/here/xyz/hub/AbstractHttpServerVerticle.java

@@ -155,7 +155,8 @@ protected void onResponseEnd(RoutingContext context) {
       logger.info(marker, "The request was cancelled. No response has been sent.");
       onRequestCancelled(context);
     }
-    logger.info(marker, "{}", LogUtil.responseToLogEntry(context));
+    // Commented this log as it is already getting logged below
+    // logger.info(marker, "{}", LogUtil.responseToLogEntry(context));
     LogUtil.addResponseInfo(context).end();
     LogUtil.writeAccessLog(context);
   }

xyz-hub-service/src/main/java/com/here/xyz/hub/Service.java

@@ -36,6 +36,7 @@
 import com.here.xyz.hub.rest.admin.messages.brokers.Broker;
 import com.here.xyz.hub.rest.admin.messages.brokers.NoopBroker;
 import com.here.xyz.hub.util.ARN;
+import com.here.xyz.hub.util.OTelMetrics;
 import com.here.xyz.hub.util.metrics.GcDurationMetric;
 import com.here.xyz.hub.util.metrics.GlobalInflightRequestMemory;
 import com.here.xyz.hub.util.metrics.GlobalUsedRfcConnections;
@@ -206,7 +207,13 @@ private static void onConfigLoaded(JsonObject jsonConfig) {
         });
         subscriptionConfigClient.init(subscriptionConfigReady -> {
           if (subscriptionConfigReady.succeeded()) {
-            XYZTransactionHandler.getInstance(rawConfiguration).start();
+            try {
+              XYZTransactionHandler.getInstance(rawConfiguration).start();
+            }
+            catch (Exception ex) {
+              logger.error("Failed to initiate background Transaction Handler job. ", ex);
+              die(1, "Background Transaction Handler job initialization failed", ex.getCause());
+            }
           } else {
             die(1, "Subscription config client failed", subscriptionConfigReady.cause());
           }
@@ -215,6 +222,8 @@ private static void onConfigLoaded(JsonObject jsonConfig) {
         die(1, "Space config client failed", ar.cause());
       }
     });
+    // Initialize OpenTelemetry related metrics
+    OTelMetrics.init();
   }
 
   private static void onLocalConnectorsInserted(AsyncResult<Void> result, JsonObject config) {

xyz-hub-service/src/main/java/com/here/xyz/hub/task/FeatureTaskHandler.java

@@ -773,6 +773,8 @@ private static <X extends FeatureTask> Future<Space> resolveSpace(final X task)
       if (task.space != null) //If the space is already given we don't need to retrieve it
         return Future.succeededFuture(task.space);
 
+      AbstractHttpServerVerticle.addStreamInfo(task.context, "SpaceId", task.getEvent().getSpace());
+
       //Load the space definition.
       return Space.resolveSpace(task.getMarker(), task.getEvent().getSpace())
           .compose(

xyz-hub-service/src/main/java/com/here/xyz/hub/util/OTelMetrics.java

@@ -0,0 +1,38 @@
+package com.here.xyz.hub.util;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.metrics.Meter;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+
+public class OTelMetrics {
+
+    private static final Logger logger = LogManager.getLogger();
+    private static final Meter meter = GlobalOpenTelemetry.meterBuilder("io.opentelemetry.metrics.memory")
+            .setInstrumentationVersion("1.27.0") // as per otel.version in pom.xml
+            .build();
+
+    public static void init() {
+        // This will keep collecting memory utilization in background
+        meter.gaugeBuilder("mem_used_pct")
+                .setDescription("Heap-Memory used percentage")
+                .setUnit("percent")
+                .buildWithCallback(
+                        (r) -> {
+                            Runtime rt = Runtime.getRuntime();
+                            long max = rt.maxMemory();
+                            long total = rt.totalMemory();
+                            long free = rt.freeMemory();
+                            long used = total - free;
+                            double usedPct = ((double)used/max)*100.00;
+                            BigDecimal bd = new BigDecimal(usedPct).setScale(2, RoundingMode.HALF_EVEN);
+                            r.record(bd.doubleValue());
+                        }
+                );
+    }
+
+
+}

xyz-hub-service/src/main/java/com/here/xyz/hub/util/logging/LogUtil.java

@@ -115,7 +115,11 @@ public static void addRequestInfo(RoutingContext context) {
     AccessLog accessLog = Api.Context.getAccessLog(context);
     HttpMethod method = context.request().method();
     accessLog.reqInfo.method = context.request().method().name();
-    accessLog.reqInfo.uri = context.request().uri();
+    // Remove access_token part from uri for security concerns
+    //accessLog.reqInfo.uri = context.request().uri();
+    final String uri = context.request().uri();
+    final int endPos = uri.indexOf("?");
+    accessLog.reqInfo.uri = (endPos > 0) ? uri.substring(0, endPos) : uri;
     accessLog.reqInfo.referer = context.request().getHeader(REFERER);
     accessLog.reqInfo.origin = context.request().getHeader(ORIGIN);
     if (POST.equals(method) || PUT.equals(method) || PATCH.equals(method)) {
@@ -167,5 +171,15 @@ public static void writeAccessLog(RoutingContext context) {
 
     accessLog.streamId = marker.getName();
     logger.log(STREAM_LEVEL, ACCESS_LOG_MARKER, accessLog.serialize());
+
+    // Log relevant details for generating API metrics
+    final AccessLog.RequestInfo req = accessLog.reqInfo;
+    final AccessLog.ResponseInfo res = accessLog.respInfo;
+    final Map<String, Object> si = accessLog.streamInfo;
+    logger.info("REST API stats for [streamId={}] [format => eventType,spaceId,connectorId,method,uri,status,timeTakenMs,resSize] - RESTAPIStats {} {} {} {} {} {} {}",
+            accessLog.streamId,
+            (si == null) ? "-" : si.get("SpaceId"),
+            (si == null) ? "-" : si.get("SID"),
+            req.method, req.uri, res.statusCode, accessLog.ms, res.size);
   }
 }

xyz-hub-service/src/main/resources/config.json

@@ -59,8 +59,8 @@
   "TXN_PUB_TPOOL_MAX_SIZE" : 100,
   "TXN_PUB_TPOOL_KEEP_ALIVE_SEC" : 60,
 
-  "AWS_ACCESS_KEY_ID": "<<some-valid-value-for-publisher>>",
-  "AWS_SECRET_ACCESS_KEY": "<<some-valid-value-for-publisher>>",
+  "AWS_ACCESS_KEY_ID": "",
+  "AWS_SECRET_ACCESS_KEY": "",
   "AWS_DEFAULT_REGION": "us-east-1",
 
   "ENABLE_TXN_SEQUENCER" : true,

xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseHandler.java

@@ -644,7 +644,7 @@ protected XyzResponse executeModifyFeatures(ModifyFeaturesEvent event) throws Ex
                         throw e;//Table does not exist yet - create it!
                     else {
 
-                        logger.warn("{} Transaction has failed. {]", traceItem, e);
+                        logger.warn("{} Transaction has failed. ", traceItem, e);
                         connection.close();
 
                         Map<String, Object> errorDetails = new HashMap<>();

xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseStreamWriter.java

@@ -114,8 +114,8 @@ protected static FeatureCollection insertFeatures(DatabaseHandler dbh, String sc
         insertStmt.close();
         insertWithoutGeometryStmt.close();
         final long duration = System.currentTimeMillis() - startTS;
-        logger.info("{} NonTransactional DB Operation Stats [format => eventType,opType,timeTakenMs] - {} {} {}",
-                traceItem, "DBOperationStats", TYPE_INSERT, duration);
+        logger.info("{} NonTransactional DB Operation Stats [format => eventType,table,opType,timeTakenMs] - {} {} {} {}",
+                traceItem, "DBOperationStats", table, TYPE_INSERT, duration);
 
         return collection;
     }
@@ -215,8 +215,8 @@ protected static FeatureCollection updateFeatures(DatabaseHandler dbh, String sc
         updateStmt.close();
         updateWithoutGeometryStmt.close();
         final long duration = System.currentTimeMillis() - startTS;
-        logger.info("{} NonTransactional DB Operation Stats [format => eventType,opType,timeTakenMs] - {} {} {}",
-                traceItem, "DBOperationStats", TYPE_UPDATE, duration);
+        logger.info("{} NonTransactional DB Operation Stats [format => eventType,table,opType,timeTakenMs] - {} {} {} {}",
+                traceItem, "DBOperationStats", table, TYPE_UPDATE, duration);
 
         return collection;
     }
@@ -273,7 +273,7 @@ protected static void deleteFeatures( DatabaseHandler dbh, String schema, String
         deleteStmt.close();
         deleteStmtWithoutUUID.close();
         final long duration = System.currentTimeMillis() - startTS;
-        logger.info("{} NonTransactional DB Operation Stats [format => eventType,opType,timeTakenMs] - {} {} {}",
-                traceItem, "DBOperationStats", TYPE_DELETE, duration);
+        logger.info("{} NonTransactional DB Operation Stats [format => eventType,table,opType,timeTakenMs] - {} {} {} {}",
+                traceItem, "DBOperationStats", table, TYPE_DELETE, duration);
     }
 }

xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseTransactionalWriter.java

@@ -94,7 +94,7 @@ public static FeatureCollection insertFeatures(DatabaseHandler dbh, String schem
         }
 
         executeBatchesAndCheckOnFailures(dbh, insertIdList, insertWithoutGeometryIdList,
-                insertStmt, insertWithoutGeometryStmt, featureList, featureWithoutGeoList, fails, false, TYPE_INSERT, traceItem);
+                insertStmt, insertWithoutGeometryStmt, featureList, featureWithoutGeoList, fails, false, TYPE_INSERT, traceItem, table);
 
         if(fails.size() > 0) {
             logException(null, traceItem, LOG_EXCEPTION_INSERT, table);
@@ -177,7 +177,7 @@ public static FeatureCollection updateFeatures(DatabaseHandler dbh, String schem
         }
 
         executeBatchesAndCheckOnFailures(dbh, updateIdList, updateWithoutGeometryIdList,
-                updateStmt, updateWithoutGeometryStmt, featureList, featureWithoutGeoList, fails, handleUUID, TYPE_UPDATE, traceItem);
+                updateStmt, updateWithoutGeometryStmt, featureList, featureWithoutGeoList, fails, handleUUID, TYPE_UPDATE, traceItem, table);
 
         if(fails.size() > 0) {
             logException(null, traceItem, LOG_EXCEPTION_UPDATE, table);
@@ -250,7 +250,7 @@ protected static void deleteFeatures(DatabaseHandler dbh, String schema, String
                 batchDeleteStmtVersioned.setArray(3, handleUUID ? connection.createArrayOf("text", versionedUuidList.toArray()) : null);
             }
             executeBatchesAndCheckOnFailures(dbh, deleteIdList, deleteIdListWithoutUUID,
-                    batchDeleteStmtVersioned, batchDeleteStmtVersionedWithoutUUID, null, null, fails, handleUUID, TYPE_DELETE, traceItem);
+                    batchDeleteStmtVersioned, batchDeleteStmtVersionedWithoutUUID, null, null, fails, handleUUID, TYPE_DELETE, traceItem, table);
 
         }else{
             if (deleteIdListWithoutUUID.size() > 0) {
@@ -261,7 +261,7 @@ protected static void deleteFeatures(DatabaseHandler dbh, String schema, String
                 batchDeleteStmt.setArray(2, handleUUID ? connection.createArrayOf("text", deleteUuidList.toArray()) : null);
             }
             executeBatchesAndCheckOnFailures(dbh, deleteIdList, deleteIdListWithoutUUID,
-                batchDeleteStmt, batchDeleteStmtWithoutUUID, null, null, fails, handleUUID, TYPE_DELETE, traceItem);
+                batchDeleteStmt, batchDeleteStmtWithoutUUID, null, null, fails, handleUUID, TYPE_DELETE, traceItem, table);
         }
 
         if(fails.size() > 0) {
@@ -274,7 +274,7 @@ private static void executeBatchesAndCheckOnFailures(DatabaseHandler dbh, List<S
                                  PreparedStatement batchStmt, PreparedStatement batchStmt2,
                                  final List<Feature> featureList, final List<Feature> featureWithoutGeoList,
                                  List<FeatureCollection.ModificationFailure> fails,
-                                 boolean handleUUID, int type, TraceItem traceItem) throws SQLException, JsonProcessingException {
+                                 boolean handleUUID, int type, TraceItem traceItem, final String table) throws SQLException, JsonProcessingException {
 
         try {
             final long startTS = System.currentTimeMillis();
@@ -298,8 +298,8 @@ private static void executeBatchesAndCheckOnFailures(DatabaseHandler dbh, List<S
                 if (rs!=null) rs.close();
             }
             final long duration = System.currentTimeMillis() - startTS;
-            logger.info("{} Transactional DB Operation Stats [format => eventType,opType,timeTakenMs] - {} {} {}",
-                    traceItem, "DBOperationStats", type, duration);
+            logger.info("{} Transactional DB Operation Stats [format => eventType,table,opType,timeTakenMs] - {} {} {} {}",
+                    traceItem, "DBOperationStats", table, type, duration);
         }finally {
             batchStmt.close();
             batchStmt2.close();