[FSTORE-1580] OnlineFS Observability

java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamProducer.java

@@ -19,6 +19,8 @@
 
 import com.logicalclocks.hsfs.FeatureStoreException;
 import com.logicalclocks.hsfs.beam.StreamFeatureGroup;
+import com.logicalclocks.hsfs.engine.FeatureGroupUtils;
+
 import lombok.NonNull;
 
 import org.apache.avro.Schema;
@@ -46,13 +48,11 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -64,7 +64,7 @@ public class BeamProducer extends PTransform<@NonNull PCollection<Row>, @NonNull
   private transient Schema encodedSchema;
   private Map<String, Schema> deserializedComplexFeatureSchemas;
   private List<String> primaryKeys;
-  private final Map<String, byte[]> headerMap = new HashMap<>();
+  private final Map<String, byte[]> headerMap;
 
   public BeamProducer(String topic, Map<String, String> properties, Schema schema, Schema encodedSchema,
                       Map<String, Schema> deserializedComplexFeatureSchemas, List<String> primaryKeys,
@@ -75,12 +75,7 @@ public BeamProducer(String topic, Map<String, String> properties, Schema schema,
     this.properties = properties;
     this.deserializedComplexFeatureSchemas = deserializedComplexFeatureSchemas;
     this.primaryKeys = primaryKeys;
-
-    headerMap.put("projectId",
-        String.valueOf(streamFeatureGroup.getFeatureStore().getProjectId()).getBytes(StandardCharsets.UTF_8));
-    headerMap.put("featureGroupId", String.valueOf(streamFeatureGroup.getId()).getBytes(StandardCharsets.UTF_8));
-    headerMap.put("subjectId",
-        String.valueOf(streamFeatureGroup.getSubject().getId()).getBytes(StandardCharsets.UTF_8));
+    this.headerMap = FeatureGroupUtils.getHeaders(streamFeatureGroup, null);
   }
 
   @Override

java/flink/src/main/java/com/logicalclocks/hsfs/flink/engine/KafkaRecordSerializer.java

@@ -18,7 +18,9 @@
 package com.logicalclocks.hsfs.flink.engine;
 
 import com.logicalclocks.hsfs.FeatureStoreException;
+import com.logicalclocks.hsfs.engine.FeatureGroupUtils;
 import com.logicalclocks.hsfs.flink.StreamFeatureGroup;
+
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.BinaryEncoder;
 import org.apache.avro.io.DatumWriter;
@@ -32,25 +34,19 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 public class KafkaRecordSerializer implements KafkaRecordSerializationSchema<GenericRecord> {
 
   private final String topic;
   private final List<String> primaryKeys;
-  private final Map<String, byte[]> headerMap = new HashMap<>();
+  private final Map<String, byte[]> headerMap;
 
   KafkaRecordSerializer(StreamFeatureGroup streamFeatureGroup) throws FeatureStoreException, IOException {
     this.topic = streamFeatureGroup.getOnlineTopicName();
     this.primaryKeys = streamFeatureGroup.getPrimaryKeys();
-
-    headerMap.put("projectId",
-        String.valueOf(streamFeatureGroup.getFeatureStore().getProjectId()).getBytes(StandardCharsets.UTF_8));
-    headerMap.put("featureGroupId", String.valueOf(streamFeatureGroup.getId()).getBytes(StandardCharsets.UTF_8));
-    headerMap.put("subjectId",
-        String.valueOf(streamFeatureGroup.getSubject().getId()).getBytes(StandardCharsets.UTF_8));
+    this.headerMap = FeatureGroupUtils.getHeaders(streamFeatureGroup, null);
   }
 
   @Override

java/hsfs/src/main/java/com/logicalclocks/hsfs/FeatureGroupBase.java

@@ -22,6 +22,7 @@
 import com.logicalclocks.hsfs.constructor.QueryBase;
 import com.logicalclocks.hsfs.engine.FeatureGroupEngineBase;
 import com.logicalclocks.hsfs.engine.FeatureGroupUtils;
+import com.logicalclocks.hsfs.metadata.OnlineIngestionApi;
 import com.logicalclocks.hsfs.metadata.Statistics;
 import com.logicalclocks.hsfs.metadata.Subject;
 import com.logicalclocks.hsfs.metadata.User;
@@ -153,6 +154,7 @@ public abstract class FeatureGroupBase<T> {
 
   protected FeatureGroupEngineBase featureGroupEngineBase = new FeatureGroupEngineBase();
   protected FeatureGroupUtils utils = new FeatureGroupUtils();
+  protected OnlineIngestionApi onlineIngestionApi = new OnlineIngestionApi();
 
   protected static final Logger LOGGER = LoggerFactory.getLogger(FeatureGroupBase.class);
 
@@ -543,5 +545,14 @@ public Schema getDeserializedAvroSchema() throws FeatureStoreException, IOExcept
     return utils.getDeserializedAvroSchema(getAvroSchema());
   }
 
+  @JsonIgnore
+  public OnlineIngestion getLatestOnlineIngestion() throws FeatureStoreException, IOException {
+    return onlineIngestionApi.getOnlineIngestion(this, "filter_by=LATEST").get(0);
+  }
+
+  @JsonIgnore
+  public OnlineIngestion getOnlineIngestion(Integer id) throws FeatureStoreException, IOException {
+    return onlineIngestionApi.getOnlineIngestion(this, "filter_by=ID:" + id).get(0);
+  }
 
 }

java/hsfs/src/main/java/com/logicalclocks/hsfs/OnlineIngestion.java

@@ -0,0 +1,101 @@
+/*
+ *  Copyright (c) 2025. Hopsworks AB
+ *
+ *  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.
+ *
+ */
+
+package com.logicalclocks.hsfs;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.logicalclocks.hsfs.metadata.RestDto;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@NoArgsConstructor
+@AllArgsConstructor
+public class OnlineIngestion extends RestDto<OnlineIngestion> {
+
+  protected static final Logger LOGGER = LoggerFactory.getLogger(OnlineIngestion.class);
+
+  @Getter
+  @Setter
+  private Integer id;
+
+  @Getter
+  @Setter
+  private Long numEntries;
+
+  @Getter
+  @Setter
+  private List<OnlineIngestionResult> results;
+
+  @Getter
+  @Setter
+  private FeatureGroupBase featureGroup;
+
+  public OnlineIngestion(long numEntries) {
+    this.numEntries = numEntries;
+  }
+
+  public void refresh() throws FeatureStoreException, IOException {
+    OnlineIngestion onlineIngestion = featureGroup.getOnlineIngestion(id);
+
+    // Method to copy data from another object
+    this.id = onlineIngestion.id;
+    this.numEntries = onlineIngestion.numEntries;
+    this.results = onlineIngestion.results;
+    this.featureGroup = onlineIngestion.featureGroup;
+  }
+
+  public void waitForCompletion(int timeout, int period)
+      throws InterruptedException, FeatureStoreException, IOException {
+    long startTime = System.currentTimeMillis();
+
+    // Convert to milliseconds
+    timeout = timeout * 1000;
+    period = period * 1000;
+
+    while (true) {
+      // Get total number of rows processed
+      long rowsProcessed = results.stream().collect(Collectors.summarizingLong(o -> o.getRows())).getSum();
+
+      // Check if the online ingestion is complete
+      if (numEntries != null && rowsProcessed >= numEntries) {
+        break;
+      }
+
+      // Check if the timeout has been reached (if timeout is 0 we will wait indefinitely)
+      if (timeout != 0 && System.currentTimeMillis() - startTime > timeout) {
+        LOGGER.warn("Timeout of " + timeout
+            + " was exceeded while waiting for online ingestion completion.");
+        break;
+      }
+
+      // Sleep for the specified period in seconds
+      Thread.sleep(period);
+
+      refresh();
+    }
+  }
+
+}

java/hsfs/src/main/java/com/logicalclocks/hsfs/OnlineIngestionResult.java

@@ -0,0 +1,41 @@
+/*
+ *  Copyright (c) 2025. Hopsworks AB
+ *
+ *  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.
+ *
+ */
+
+package com.logicalclocks.hsfs;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@NoArgsConstructor
+@AllArgsConstructor
+public class OnlineIngestionResult {
+
+  @Getter
+  @Setter
+  private Integer onlineIngestionId;
+
+  @Getter
+  @Setter
+  private String status;
+
+  @Getter
+  @Setter
+  private Long rows;
+
+}

java/hsfs/src/main/java/com/logicalclocks/hsfs/engine/FeatureGroupUtils.java

@@ -20,11 +20,13 @@
 import com.logicalclocks.hsfs.metadata.FeatureGroupApi;
 import com.logicalclocks.hsfs.metadata.HopsworksClient;
 import com.logicalclocks.hsfs.metadata.KafkaApi;
+import com.logicalclocks.hsfs.metadata.OnlineIngestionApi;
 import com.logicalclocks.hsfs.metadata.Subject;
 import com.logicalclocks.hsfs.Feature;
 import com.logicalclocks.hsfs.FeatureGroupBase;
 import com.logicalclocks.hsfs.FeatureGroupCommit;
 import com.logicalclocks.hsfs.FeatureStoreException;
+import com.logicalclocks.hsfs.OnlineIngestion;
 
 import lombok.SneakyThrows;
 import org.apache.avro.Schema;
@@ -34,6 +36,7 @@
 import scala.collection.Seq;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.sql.Timestamp;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -247,4 +250,25 @@ public String getDatasetType(String path) {
     }
     return "DATASET";
   }
+
+  public static Map<String, byte[]> getHeaders(FeatureGroupBase featureGroup, Long numEntries)
+      throws FeatureStoreException, IOException {
+    Map<String, byte[]> headerMap = new HashMap<>();
+
+    headerMap.put("projectId",
+        String.valueOf(featureGroup.getFeatureStore().getProjectId()).getBytes(StandardCharsets.UTF_8));
+    headerMap.put("featureGroupId", String.valueOf(featureGroup.getId()).getBytes(StandardCharsets.UTF_8));
+    headerMap.put("subjectId",
+        String.valueOf(featureGroup.getSubject().getId()).getBytes(StandardCharsets.UTF_8));
+
+    if (featureGroup.getOnlineEnabled()) {
+      OnlineIngestion onlineIngestion = new OnlineIngestionApi()
+          .createOnlineIngestion(featureGroup, new OnlineIngestion(numEntries));
+
+      headerMap.put("onlineIngestionId",
+          String.valueOf(onlineIngestion.getId()).getBytes(StandardCharsets.UTF_8));
+    }
+
+    return headerMap;
+  }
 }

java/hsfs/src/main/java/com/logicalclocks/hsfs/metadata/OnlineIngestionApi.java

@@ -0,0 +1,85 @@
+/*
+ *  Copyright (c) 2025. Hopsworks AB
+ *
+ *  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.
+ *
+ */
+
+package com.logicalclocks.hsfs.metadata;
+
+import com.damnhandy.uri.template.UriTemplate;
+import com.logicalclocks.hsfs.FeatureGroupBase;
+import com.logicalclocks.hsfs.FeatureStoreException;
+import com.logicalclocks.hsfs.OnlineIngestion;
+
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+public class OnlineIngestionApi {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(OnlineIngestionApi.class);
+
+  public static final String ONLINE_INGESTION_PATH = "/featuregroups/{fgId}/online_ingestion";
+
+  public OnlineIngestion createOnlineIngestion(FeatureGroupBase featureGroup, OnlineIngestion onlineIngestion)
+      throws FeatureStoreException, IOException {
+    HopsworksClient hopsworksClient = HopsworksClient.getInstance();
+    String pathTemplate = HopsworksClient.PROJECT_PATH
+        + FeatureStoreApi.FEATURE_STORE_PATH
+        + ONLINE_INGESTION_PATH;
+
+    String uri = UriTemplate.fromTemplate(pathTemplate)
+        .set("projectId", hopsworksClient.getProject().getProjectId())
+        .set("fsId", featureGroup.getFeatureStore().getId())
+        .set("fgId", featureGroup.getId())
+        .expand();
+
+    HttpPost postRequest = new HttpPost(uri);
+    postRequest.setHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+    postRequest.setEntity(hopsworksClient.buildStringEntity(onlineIngestion));
+
+    LOGGER.info("Sending metadata request: " + uri);
+
+    onlineIngestion = hopsworksClient.handleRequest(postRequest, OnlineIngestion.class);
+    onlineIngestion.setFeatureGroup(featureGroup);
+    return onlineIngestion;
+  }
+
+  public List<OnlineIngestion> getOnlineIngestion(FeatureGroupBase featureGroup, String queryParameters)
+      throws FeatureStoreException, IOException {
+    HopsworksClient hopsworksClient = HopsworksClient.getInstance();
+    String pathTemplate = HopsworksClient.PROJECT_PATH
+        + FeatureStoreApi.FEATURE_STORE_PATH
+        + ONLINE_INGESTION_PATH;
+
+    String uri = UriTemplate.fromTemplate(pathTemplate)
+        .set("projectId", hopsworksClient.getProject().getProjectId())
+        .set("fsId", featureGroup.getFeatureStore().getId())
+        .set("fgId", featureGroup.getId())
+        .expand() + "?" + queryParameters;
+
+    LOGGER.debug("Sending metadata request: " + uri);
+
+    OnlineIngestion onlineIngestion = hopsworksClient.handleRequest(new HttpGet(uri), OnlineIngestion.class);
+    for (OnlineIngestion ingestion : onlineIngestion.getItems()) {
+      ingestion.setFeatureGroup(featureGroup);
+    }
+    return onlineIngestion.getItems();
+  }
+}