Emit state of replace and append for native batch tasks #12488

loquisgon · 2022-04-28T02:35:06Z

A native batch ingestion job can be in either of these three modes: REPLACE (new since tombstones were added), APPEND, and REPLACE_LEGACY (current default for native batch ingestion). These modes are set with appropriate values for the following IOConfig flags: appendToExisting and dropExisting. The goal of this ticket is to emit Druid metrics to be able to assess the utilization of the different flavors of batch ingestion, in particular replace & tombstone creation.

Copying from the metrics.md documentation:

Metric	Description	Dimensions	Normal Value
`ingest/count`	Count of `1` every time an ingestion job runs (includes compaction jobs). Aggregate using dimensions.	dataSource, taskId, taskType, taskIngestionMode	Always `1`.
`ingest/segments/count`	Count of final segments created by job (includes tombstones).	dataSource, taskId, taskType, taskIngestionMode	At least `1`.
`ingest/tombstones/count`	Count of tombstones created by job	dataSource, taskId, taskType, taskIngestionMode	Zero or more for replace. Always zero for non-replace tasks (always zero for legacy replace, see below).

The taskIngestionMode dimension includes the following modes:
APPEND, REPLACE_LEGACY, and REPLACE. The APPEND mode indicates a native
ingestion job that is appending to existing segments; REPLACE a native ingestion
job replacing existing segments using tombstones;
and REPLACE_LEGACY the original replace before tombstones.

The mode is decided using the values
of the isAppendToExisting and isDropExisting flags in the
task's IOConfig as follows:

`isAppendToExisting`	`isDropExisting`	mode
`true`	`false`	`APPEND`
`true`	`true`	Invalid combination, exception thrown.
`false`	`false`	`REPLACE_LEGACY` (this is the default for native batch ingestion).
`false`	`true`	`REPLACE`

This PR has:

[ X] been self-reviewed.
- using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
[ X] added documentation for new or modified features or behaviors.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added or updated version, license, or notice information in licenses.yaml
added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
[ X] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
added integration tests.
[ X] been tested in a test Druid cluster.

…E, REPLACE)

zachjsh · 2022-05-10T19:28:54Z

docs/operations/metrics.md

+`true` | `false` | `APPEND`|
+`true` | `true  ` | Invalid combination, exception thrown. |
+`false` | `false` | `OVERWRITE` (this is the default for native batch ingestion). |
+`false` | `true` | `REPLACE`|


Might want to add details here about the creation of tombstones for time-chunks within the date interval specified for ingestion that have no data, if in REPLACE mode, and the absence of this in OVERWRITE mode, if my understanding is correct?

Well, this is to document the batch ingestion mode not the semantics of the method so I am not sure that the explanation for its semantics belongs here.

Oh ok, I guess the semantics are described elsewhere? If so should we link to that? not a big deal

Yes, in the documentation for the flag isDropExisting. The actual names APPEND, REPLACE, and OVERWRITE are not formally documented and I am introducing them here. But documenting them is our of scope for this ticket I feel.

Added some explanation about what the modes mean.

zachjsh · 2022-05-10T19:40:35Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java

+        emitter.emit(buildEvent("compact/overwrite/count", 1));
+        break;
+      default:
+        throw new ISE("Invalid compact ingestion mode [%s]", mode);


Should we just log error instead of possibly failing ingestion task with this thrown exception?

Integrating the BatchIngestionMode (see comment below) made this not required anymore.

zachjsh · 2022-05-10T19:42:10Z

docs/operations/metrics.md

+|`isAppendToExisting` | `isdDropExisting` | mode |
+|---------------------|-------------------|------|
+`true` | `false` | `APPEND`|
+`true` | `true  ` | Invalid combination, exception thrown. |


Looking at code, it doesnt seem that we throw an exception in this case, but rather treat it as APPEND, since we consider a job an append job if isAppendToExisting is true. and short-circuit the check to isDropExisting

Yes it does but it was confusing where. Due to this great comment I decided to fully integrate the concept of BatchIngestionMode in the batch ingestion task internals replacing all the confusing tests for append or drop existing etc. There were many changes but they should straightforward to follow. The code is better now, IMO.

suneet-s · 2022-05-11T05:51:16Z

The goal of this ticket is to emit Druid metrics to be able to assess the utilization of the different flavors of batch ingestion

Since there are many different config flags that might be of interest, and these can change as devs add new config settings. Have you considered extending the RequestLogger interface to allow logging ingestion specs in addition to the current sql and native queries? This will allow cluster operators to point these logs to another system (maybe even a druid cluster) for running analytics on top of the entire ingestion spec.

loquisgon · 2022-05-11T18:28:05Z

@suneet-s No I have not considered RequestLogger, I am not really familiar with that. But I see your point. However, the whole area of enriching, cleaning, the logging, metrics, and alerts for ingestion is an area ripe for improvement. But I would suggest to think a little about a decent starting point on what to do/unify these areas. So for now, I would like to focus on the scope as defined for the work, but I would strongly recommend a more extended approach that would incorporate suggestions like the one you are making.

suneet-s · 2022-05-11T18:50:43Z

docs/operations/metrics.md

+of the `isAppendToExisting` and `isDropExisting` flags in the
+task's `IOConfig` as follows:
+
+|`isAppendToExisting` | `isdDropExisting` | mode |


Nice catch!

fix in next commit

suneet-s · 2022-05-11T18:51:59Z

docs/operations/metrics.md

+
+|Metric|Description|Dimensions|Normal Value|
+|------|-----------|----------|------------|
+|`ingest/compact/ovewrite/count`|Count of `1` every time a compaction overwrite job runs|dataSource, taskId, taskType|Always `1`.|


Won't taskType always be compact ?

taskType yeah for compact is compact but for other batch can be other...so for completeness & consistency it is better to live that dimension there I think

Removed task type from metric.

suneet-s · 2022-05-11T18:58:35Z

docs/operations/metrics.md

@@ -221,6 +221,41 @@ Note: If the JVM does not support CPU time measurement for the current thread, i
 |`worker/taskSlot/total/count`|Number of total task slots on the reporting worker per emission period. This metric is only available if the WorkerTaskCountStatsMonitor module is included.|category, version.|Varies.|
 |`worker/taskSlot/used/count`|Number of busy task slots on the reporting worker per emission period. This metric is only available if the WorkerTaskCountStatsMonitor module is included.|category, version.|Varies.|

+## Batch ingestion metrics (Native parallel task)


Why did we decide to have a separate metric name for each type vs a single metric name with the type being a dimension?

ingest/batch/count, ingest/batch/segments/count and a dimension type (or even 2 dimensions for isAppendToExisting and isDropExisting - in this case, you don't need to deal with the error case described on line 243)

`true` | `true ` | Invalid combination, exception thrown. |

In a similar vein, won't these metrics be useful for streaming ingestion as well - Why not make the metric name ingest/count and then the taskType dimension tells you whether or not it is streaming or batch?

Also, reading these docs, it's not clear to me whether a compaction job will also be counted as a batch job

The goal of this feature is to gather data to let us know the utilization of the various modes of batch ingest, in particular REPLACE thus the categorization.

The invalid combination is enabled by the fact that the IOConfig part of the spec exposes both dropExisting and appendToExisting today. If we were to deprecate those and expose instead batchProcessingMode with its three possible values then this invalid state would no longer be possible. But I think that it is better to do this later if we agree it is a good idea.

The taskType is already a dimension (i.e. for native batch types we have: compact, index, index_parallel and for streaming index_kafka , etc; for non native index_hadoop...) so yeah...maybe I should just drop the batch (and compact) from the metric name and just use the indexType dimension to filter...is this what you are thinking @suneet-s ?

I agree with @suneet-s suggestion to drop the batch and compact from metric name, as the task-type dimension included signifies this, and then the metric can be reused for other ingestion types, not just batch / compact.

Yes, I agree too. Working on it now.

In a similar vein, won't these metrics be useful for streaming ingestion as well - Why not make the metric name ingest/count and then the taskType dimension tells you whether or not it is streaming or batch?

I had suggested a batch-specific name when discussing with Agustin since I think the semantics of the metric are batch specific, it's being emitted once per ingest spec posted, and I'm not sure what the equivalent for streaming would be. It's a metric indicating "number of times user did some action with a specific intent" vs. "detail of execution" (like a parallel subtask count for batch or number of streaming subtasks). Taking it out for a shared metric name seems fine to me though.

@suneet-s I have removed the batch type from the metric and added taskIngestionMode as a dimension. As a result the new metrics are much clean and the code is cleaner as well. Also, now it was simple to add the metrics to streaming as well, take a look.

zachjsh · 2022-05-12T16:26:44Z

docs/operations/metrics.md

+|`ingest/compact/ovewrite/count`|Count of `1` every time a compaction overwrite job runs|dataSource, taskId, taskType|Always `1`.|
+|`ingest/compact/replace/count`|Count of `1` every time a compaction replace job runs|dataSource, taskId, taskType|Always `1`.|
+
+|`isDropExisting` | mode |


Should this table have a description? Or maybe I missed it

@zachjsh Simplified & cleaned up docs... see PR description or the metrics.md changes

…nsion. Move IngestionMode to AbstractTask to facilitate having mode as a dimension. Add metrics to streaming. Add missing coverage.

lgtm-com · 2022-05-18T05:22:50Z

This pull request introduces 2 alerts when merging 1622b74 into 3e8d7a6 - view on LGTM.com

new alerts:

2 for Dereferenced variable may be null

…ation of IngestionMode to make it more robust to null IOConfig and fix test.

lgtm-com · 2022-05-18T19:37:49Z

This pull request introduces 1 alert when merging c5f139f into 3e8d7a6 - view on LGTM.com

new alerts:

1 for Dereferenced variable may be null

loquisgon · 2022-05-18T19:42:02Z

I don't understand the LGTM concern about the "new publishedSegmentsAndCommitMetadata". That variable existed already before I added the metric. Maybe it is complaining because the new code explicitly check that the variable is not null?

lgtm-com · 2022-05-18T23:24:40Z

This pull request introduces 1 alert when merging 8ba3f60 into 3e8d7a6 - view on LGTM.com

new alerts:

1 for Dereferenced variable may be null

jon-wei · 2022-05-20T20:49:02Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java

+      isAppendToExisting = BatchIOConfig.DEFAULT_APPEND_EXISTING;
+      isDropExisting = BatchIOConfig.DEFAULT_DROP_EXISTING;
+    } else {
+      isAppendToExisting = BatchIOConfig.DEFAULT_APPEND_EXISTING;


How come this uses the default from BatchIOConfig instead of using ioConfig.isAppendToExisting()

This is specific to CompactionIOCOnfig which does not have a isAppendToExisting() flag/method. Added a comment to the code so this is more clear.

jon-wei · 2022-05-20T20:49:32Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java

+    return ingestionMode;
+  }
+
+  protected static IngestionMode computeIngestionMode(@Nullable CompactionIOConfig ioConfig)


Suggest distinguishing these two computeIngestionMode methods with different names since they both have nullable parameters

Good call. Done in next commit.

lgtm-com · 2022-05-20T21:40:04Z

This pull request introduces 1 alert when merging 2945222 into c236227 - view on LGTM.com

new alerts:

1 for Dereferenced variable may be null

…or if they are passed null. Other minor cleanup.

lgtm-com · 2022-05-21T00:13:54Z

This pull request introduces 1 alert when merging 59aa6d7 into c236227 - view on LGTM.com

new alerts:

1 for Dereferenced variable may be null

Emit state of replace and append for native batch tasks

f76f83b

loquisgon marked this pull request as draft April 28, 2022 18:05

Agustin Gonzalez added 2 commits April 28, 2022 14:43

Emit count of one depending on batch ingestion mode (APPEND, OVERWRIT…

fb87412

…E, REPLACE)

Add metric to compaction job

c54bce7

loquisgon marked this pull request as ready for review April 29, 2022 01:34

Agustin Gonzalez added 7 commits April 28, 2022 18:46

Avoid null ptr exc when null emitter

4cd4843

Coverage

252641e

Emit tombstone & segment counts

87435a2

Tasks need a type

05eb53b

Merge branch 'master' into dropexisting_metric

f02fb4a

Spelling

ac26081

Merge branch 'master' into dropexisting_metric

dd545a7

FrankChen021 added the Area - Metrics/Event Emitting label May 4, 2022

zachjsh reviewed May 10, 2022

View reviewed changes

Integrate BatchIngestionMode in batch ingestion tasks functionality

c543dba

jon-wei added the Release Notes label May 11, 2022

jon-wei approved these changes May 11, 2022

View reviewed changes

suneet-s reviewed May 11, 2022

View reviewed changes

suneet-s added the Design Review label May 11, 2022

Typos

bbcacf2

zachjsh reviewed May 12, 2022

View reviewed changes

Remove batch ingestion type from metric since it is already in a dime…

1622b74

…nsion. Move IngestionMode to AbstractTask to facilitate having mode as a dimension. Add metrics to streaming. Add missing coverage.

Avoid inner class referenced by sub-class inspection. Refactor comput…

c5f139f

…ation of IngestionMode to make it more robust to null IOConfig and fix test.

Spelling

8ba3f60

loquisgon mentioned this pull request May 18, 2022

Deal with potential cardinality estimate being negative and add logging to hash determine partitions phase #12443

Merged

5 tasks

Avoid polluting the Task interface

2945222

jon-wei reviewed May 20, 2022

View reviewed changes

Rename computeCompaction methods to avoid ambiguous java compiler err…

59aa6d7

…or if they are passed null. Other minor cleanup.

loquisgon merged commit 2f3d7a4 into apache:master May 23, 2022

loquisgon deleted the dropexisting_metric branch May 23, 2022 19:32

abhishekagarwal87 added this to the 24.0.0 milestone Aug 26, 2022

techdocsmith mentioned this pull request Aug 26, 2022

[Draft] 24.0 Release notes #12825

Closed

abhishekagarwal87 mentioned this pull request Sep 8, 2022

Test issue [Please ignore] #13055

Closed

	\|`isAppendToExisting` \| `isdDropExisting` \| mode \|
	\|`isAppendToExisting` \| `isDropExisting` \| mode \|

Emit state of replace and append for native batch tasks #12488

Emit state of replace and append for native batch tasks #12488

Conversation

loquisgon commented Apr 28, 2022 • edited

zachjsh May 10, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zachjsh May 11, 2022 • edited

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

loquisgon May 10, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

suneet-s commented May 11, 2022 • edited

loquisgon commented May 11, 2022

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

loquisgon May 11, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jon-wei May 17, 2022 • edited

Choose a reason for hiding this comment

loquisgon May 18, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lgtm-com bot commented May 18, 2022

lgtm-com bot commented May 18, 2022

loquisgon commented May 18, 2022

lgtm-com bot commented May 18, 2022

Choose a reason for hiding this comment

loquisgon May 20, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lgtm-com bot commented May 20, 2022

lgtm-com bot commented May 21, 2022

loquisgon commented Apr 28, 2022 •

edited

zachjsh May 10, 2022 •

edited

zachjsh May 11, 2022 •

edited

loquisgon May 10, 2022 •

edited

suneet-s commented May 11, 2022 •

edited

loquisgon May 11, 2022 •

edited

jon-wei May 17, 2022 •

edited

loquisgon May 18, 2022 •

edited

loquisgon May 20, 2022 •

edited