Add operation run store methods #15320

samdgupi · 2023-09-13T18:00:47Z

Added

create run
update run status
update run metadata
fail run
get run
scan run

samdgupi · 2023-09-29T17:35:29Z

cdap-storage-spi/src/main/java/io/cdap/cdap/spi/data/StructuredTable.java

+   *     indexed column, or the type does not match the schema
+   * @throws IOException if there is an error scanning the table
+   */
+  default CloseableIterator<StructuredRow> scan(Range keyRange, int limit,


Currently only implemented in PostgresSql. Will handle nosql in a seperate PR

this isn't needed anymore right?

...abric/src/main/java/io/cdap/cdap/internal/operations/OperationRunAlreadyExistsException.java

albertshau · 2023-10-03T22:47:06Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+
+  @VisibleForTesting
+  // USE ONLY IN TESTS: WILL DELETE ALL OPERATION RUNS
+  public void deleteAllTables() throws IOException {


methods like this should be package private

a better name would be clear() or clearData(). This name makes it sound like the table will be dropped.

albertshau · 2023-10-03T22:51:41Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+    );
+  }
+
+  private long getStartTime(OperationRunId runId)


Methods like this can lead to unnecessary DB reads over time, as callers may do a normal read and then call this method. We should remove this and force callers to get the start time from the row.

Done. Moved the logic directly to caller as there is only one usage

albertshau · 2023-10-03T22:53:31Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+    return fields;
+  }
+
+  private OperationRunDetail<?> getCurrentRunDetail(OperationRunId runId)


this should be getRunDetail. The current makes it sound like there could be a previous run with the same id.

Also, I'm not sure how this would work with the <?>. What does it get deserialized to?

The generic type T refers to the type of operation request object. It is not needed for most of the cases when we retrieve the OperationRunDetail except when we plan to launch the operation. The current approach I am taking is based on the operation type the request object type will be determined and the OperationRunDetail will be casted accordingly.

I'm still confused how this will work, doesn't the type need to be passed into gson in order for it to be deserialized properly? If you just deserialize it here as ?, don't you get a Map or a JsonObject or something as the field? I don't think you'd be able to cast it directly.

albertshau · 2023-10-03T22:54:43Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+
+  private Collection<Field<?>> getFilterIndexes(OperationRunFilter filter) {
+    Collection<Field<?>> filterIndexes = new ArrayList<>();
+    if (filter == null) {


don't need to handle his case, we should not allow the filter instance to be null

filter null represents no filter applied. I have added a new method in OperationRunFilter to get a empty filter to avoid a null check

albertshau · 2023-10-03T23:14:14Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+   * @throws OperationRunNotFoundException run with id does not exist in namespace
+   */
+  public OperationRunDetail<?> getOperation(OperationRunId runId)
+      throws NotFoundException, IOException {


throws OperationRunNotFoundException

albertshau · 2023-10-03T23:17:00Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+        Fields.stringField(StoreDefinition.OperationRunsStore.ID_FIELD, runId.getRun()));
+    fields.add(
+        Fields.longField(StoreDefinition.OperationRunsStore.UPDATE_TIME_FIELD,
+            System.currentTimeMillis()));


instead of System.currentTimeMillis(), it would be good to take a Clock in the constructor of this class and use that instead. It would make unit testing simpler, as you can expect a concrete time.

albertshau · 2023-10-03T23:19:24Z

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/data/sql/PostgreSqlStructuredTable.java

+   * @param limit number of result
+   * @return a select query
+   */
+  private PreparedStatement prepareMultiScanQuery(Collection<Range> singletonRanges,


are there any logical changes being made here or is it just getting moved? (In the future, keep methods in the same location so it is easier to review the diff)

This was done to fix a checkstyle warning about keeping all the override methods together. There is no logical change. I will revert it back to reduce the diff

I see, if it is fixing checkstyle it's ok to move it.

albertshau · 2023-10-03T23:19:52Z

cdap-proto/src/main/java/io/cdap/cdap/proto/id/OperationRunId.java

+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package io.cdap.cdap.proto.id;


fix checkstyle

albertshau · 2023-10-03T23:20:01Z

cdap-proto/src/main/java/io/cdap/cdap/proto/id/OperationRunId.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright © 2015-2019 Cask Data, Inc.


fix copyright year

samdgupi · 2023-10-04T17:41:13Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+   * @param txBatchSize batch size of transaction
+   * @param consumer {@link Consumer} to process each scanned run
+   */
+  public boolean scanOperations(ScanOperationRunsRequest request, int txBatchSize,


Unit test will be added in a followup PR. Will add a TODO

albertshau · 2023-10-06T03:16:41Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+
+    AtomicReference<ScanOperationRunsRequest> requestRef = new AtomicReference<>(request);
+    AtomicReference<String> lastKey = new AtomicReference<>(request.getScanAfter());
+    AtomicInteger currentLimit = new AtomicInteger(request.getLimit());


this looks like it can be a normal int

albertshau · 2023-10-06T03:17:18Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+   *
+   * @param request scan request including filters and limit
+   * @param txBatchSize batch size of transaction
+   * @param consumer {@link Consumer} to process each scanned run


javadoc should have @return as well

though returning whether the limit was reached is kind of a weird thing to return, not sure how it would be used. I would either make it void or return the total number of operations scanned.

The boolean would be used in the handler to find if all the operations are scanned and send last operation id accordingly. We can use the total number of operation scanned also but wanted to keep this analogous to the application scan

see:

cdap/cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/AppLifecycleHttpHandler.java

Line 307 in 550a8f3

return !pageLimitReached || record == null ? null :

hm I see, that's ok then

albertshau · 2023-10-06T03:24:52Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+      requestRef.set(batchRequest);
+
+      TransactionRunners.run(transactionRunner, context -> {
+        lastKey.set(


you can use the run method with a return value instead of setting lastKey in the closure. Then it can be a normal String instead of an AtomicReference.

albertshau · 2023-10-06T03:26:30Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+
+  private static final String SMALLEST_POSSIBLE_STRING = "";
+
+  @Inject


don't think we would use injection to create the store

albertshau · 2023-10-06T03:27:23Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+  }
+
+  @VisibleForTesting
+    // USE ONLY IN TESTS: WILL DELETE ALL OPERATION RUNS


albertshau · 2023-10-06T03:28:41Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+    Range range = Range.create(endFields, endBound, startFields, startBound);
+    Collection<Field<?>> filterIndexes = getFilterIndexes(request.getFilter());
+    StructuredTable table = getOperationRunsTable(context);
+    AtomicReference<String> lastKey = new AtomicReference<>();


this can be a normal String

albertshau · 2023-10-06T03:30:26Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/ScanOperationRunsRequest.java

+  /**
+   * namespace to return applications for or null for all namespaces.
+   */
+  @Nullable


this should not be nullable

albertshau · 2023-10-06T03:30:36Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/ScanOperationRunsRequest.java

+  }
+
+  /**
+   * namespace to return applications for or null for all namespaces.


update javadoc to not include null

albertshau · 2023-10-06T03:31:53Z

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/operations/OperationRunsStoreTest.java

+  public void testScanOperation() throws Exception {
+    insertTestRuns();
+
+    // TODO(samik) verify the actual list


should do this now rather than later

albertshau · 2023-10-06T03:45:15Z

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/data/sql/PostgreSqlStructuredTable.java

+          "is not an indexed column or primary key");
+    }
+    return scan(keyRange, limit, filterIndexes, Collections.singleton(orderByField), sortOrder,
+        true);


It is really confusing that this scan method is an 'and' filter while the other ones are an 'or' filter. Can we just make the operation type part of the primary key? That way it can go in the key range and we can use the existing methods without adding more to StructuredTable. Would like to avoid bloating the StructuredTable class, it's already pretty confusing as-is.

FYI, whenever we go to CDAP 7.0, we would like to remove Hadoop support, which would allow us to remove the NoSql implementations which would let us use normal SQL instead of this interface.

The reason operationtype is not part of the primary key because the operation id and namespace should uniquely identify a operation. This is in line with the API design /namespace/{namespace}/operation/{id}. If we add the operationtype in primary key we have to change the API also.
The API is designed to be compatible with GCP operations api which also does not include types.
I think we do need a option to select how the filters should be joined. I can create two different top level functions for AND and OR filter.

Removed and filter and added operation type in the range key

albertshau · 2023-10-10T03:41:00Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+    }
+    return GSON.fromJson(
+        row.get().getString(StoreDefinition.OperationRunsStore.DETAILS_FIELD),
+        OperationRunDetail.class


I'm still not sure how this will work without a concrete class for the generic. Maybe it will be more clear when there is code around the caller using the result, but I get the feeling it will need to revisited.

albertshau · 2023-10-10T03:42:51Z

cdap-storage-spi/src/main/java/io/cdap/cdap/spi/data/StructuredTable.java

+   *     indexed column, or the type does not match the schema
+   * @throws IOException if there is an error scanning the table
+   */
+  default CloseableIterator<StructuredRow> scan(Range keyRange, int limit,


this isn't needed anymore right?

albertshau

lgtm assuming the new scan method is removed from StructuredTable

chtyim · 2023-10-10T18:54:28Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+  private final TransactionRunner transactionRunner;
+
+  @Inject
+  public OperationLifecycleService(TransactionRunner transactionRunner) {


Remove public for injected constructor to avoid direct instantiation.

chtyim · 2023-10-10T18:55:36Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+/**
+ * Service that manages lifecycle of Operation.
+ */
+public class OperationLifecycleService {


Usually we only name a class with Service suffix if it is a guava Service. This class seems more like a Manager?

We do have classes with suffix service like ProgramLifecycleService which is not a guava service but encapsulates all operations we can do upon an entity . I was following the apparent naming convention.

Updated to use Manager Suffix

chtyim · 2023-10-10T18:56:56Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+   */
+  public boolean scanOperations(ScanOperationRunsRequest request, int txBatchSize,
+      Consumer<OperationRunDetail<?>> consumer) throws OperationRunNotFoundException, IOException {
+    AtomicReference<ScanOperationRunsRequest> requestRef = new AtomicReference<>(request);


Why we need an atomic reference? Seems like everything is happening in the local thread? A simple local variable would do the job.

chtyim · 2023-10-10T18:58:55Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+      }
+      currentLimit -= txBatchSize;
+    }
+    return currentLimit == 0;


If the above loop having lastKey == null, the while loop would break and currentLimit may not be 0. Is this expected? If the caller is also having a while loop until this method return false, would it resulted in infinite loop?

It is expected to break the loop as lastKey == null will signify no more row left to scan.
The return value here will signify if the page limit was reached. The caller would also need to track the last record scanned as the following example in AppLifeCycleHttpHandler

boolean pageLimitReached = applicationLifecycleService.scanApplications(scanRequest, appDetail -> { ApplicationRecord record = new ApplicationRecord(appDetail); jsonListResponder.send(record); lastRecord.set(record); }); ApplicationRecord record = lastRecord.get(); return !pageLimitReached || record == null ? null : record.getName() + EntityId.IDSTRING_PART_SEPARATOR + record.getAppVersion();

The caller is requierd to update the ScanAfter field in the request for the next call if this method is called from a while loop

chtyim · 2023-10-10T18:59:05Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationLifecycleService.java

+  private OperationRunsStore getOperationRunStore(StructuredTableContext context) {
+    return new OperationRunsStore(context);
+  }
+


Remove extra new lines.

chtyim · 2023-10-10T19:00:15Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunFilter.java

+/**
+ * This class defined various filters that can be applied during operation runs scanning.
+ */
+public class OperationRunFilter {


Usually a Filter is kind of a Function instead of just having getters. How is it supposed to be used?

It is just a container class to encapsulate what filter options are available for operations.

chtyim · 2023-10-10T19:00:36Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+/**
+ * Store for operation runs.
+ */
+public class OperationRunsStore {


No plural in class name.

chtyim · 2023-10-10T19:02:12Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+
+  private final Clock clock;
+
+  private static final String SMALLEST_POSSIBLE_STRING = "";


group static fields together

chtyim · 2023-10-10T19:05:40Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+
+  private void writeOperationRun(OperationRunId runId, OperationRunDetail<?> detail)
+      throws IOException {
+    Collection<Field<?>> fields = new ArrayList<>();


Consider using Arrays.asList or ImmutableList.of to improve readability.

chtyim · 2023-10-10T19:07:18Z

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/operations/OperationRunsStore.java

+  private OperationRunDetail<?> getRunDetail(OperationRunId runId)
+      throws IOException, OperationRunNotFoundException {
+    Optional<StructuredRow> row = getOperationRunInternal(runId);
+    if (!row.isPresent()) {


Avoid using isPresent on Optional. In this case, use .map and orElseThrow.

return getOperationRunInternal(runId) .map(this::rowToRunDetail) .orElseThrow(() -> new OperationRunNotFoundException(...));

Please also fix other places that use isPresent and replace with more expressive way.

Updated. There is one case where we throw error if the row is present. As we can't throw checked error from within ifPresent I have not made any change

samdgupi · 2023-10-11T13:02:06Z

@chtyim Merging this PR to unblock followup PRs for LRO. Please add comments for any suggested change and I will raise a further PR to solve.

samdgupi changed the base branch from develop to operation-store September 13, 2023 18:10

samdgupi force-pushed the operation-store branch 5 times, most recently from 5c8c266 to 1941854 Compare September 21, 2023 15:35

Base automatically changed from operation-store to develop September 22, 2023 06:54

samdgupi force-pushed the operation-store-apis branch 4 times, most recently from 93d5151 to 52f7e9d Compare September 26, 2023 07:24

samdgupi added the build Triggers github actions build label Sep 26, 2023

samdgupi force-pushed the operation-store-apis branch from 52f7e9d to 03beee6 Compare September 26, 2023 12:45

samdgupi requested review from albertshau and GnsP September 26, 2023 12:46

samdgupi marked this pull request as ready for review September 26, 2023 12:46

samdgupi force-pushed the operation-store-apis branch 2 times, most recently from 23a35de to 3d30d35 Compare September 27, 2023 06:35

samdgupi requested a review from tivv September 28, 2023 08:52

samdgupi force-pushed the operation-store-apis branch from 3d30d35 to 18300f7 Compare September 29, 2023 17:34

samdgupi commented Sep 29, 2023

View reviewed changes

albertshau reviewed Oct 3, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch 2 times, most recently from 73ae585 to ce08626 Compare October 4, 2023 17:35

samdgupi commented Oct 4, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch from ce08626 to cb66ab8 Compare October 4, 2023 17:45

samdgupi requested a review from albertshau October 5, 2023 08:21

albertshau reviewed Oct 6, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch 2 times, most recently from af6aa13 to 57a1c6e Compare October 6, 2023 16:56

samdgupi force-pushed the operation-store-apis branch 3 times, most recently from df24aec to cf0f87b Compare October 9, 2023 17:39

samdgupi requested a review from albertshau October 9, 2023 17:40

albertshau reviewed Oct 10, 2023

View reviewed changes

albertshau approved these changes Oct 10, 2023

View reviewed changes

samdgupi force-pushed the operation-store-apis branch from cf0f87b to 26ad907 Compare October 10, 2023 18:38

chtyim reviewed Oct 10, 2023

View reviewed changes

Add operation store methods

6ef10d8

samdgupi force-pushed the operation-store-apis branch from 26ad907 to 6ef10d8 Compare October 11, 2023 11:01

samdgupi merged commit 30a1b67 into develop Oct 11, 2023
5 checks passed

samdgupi deleted the operation-store-apis branch October 11, 2023 13:02

samdgupi added the LRO label for tracking Long Running Operation support label Nov 30, 2023


		private static final String SMALLEST_POSSIBLE_STRING = "";

		@Inject


		private final Clock clock;

		private static final String SMALLEST_POSSIBLE_STRING = "";

Add operation run store methods #15320

Add operation run store methods #15320

Conversation

samdgupi commented Sep 13, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samdgupi Oct 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

albertshau Oct 5, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samdgupi Oct 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

albertshau left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samdgupi Oct 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samdgupi commented Sep 13, 2023 •

edited

Loading

samdgupi Oct 10, 2023 •

edited

Loading

albertshau Oct 5, 2023 •

edited

Loading

samdgupi Oct 4, 2023 •

edited

Loading

samdgupi Oct 11, 2023 •

edited

Loading