You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2020/03/12 10:07:11 UTC
[GitHub] [druid] clintropolis opened a new pull request #9509: add kinesis
lag metric
clintropolis opened a new pull request #9509: add kinesis lag metric
URL: https://github.com/apache/druid/pull/9509
Fixes #6739.
### Description
This PR adds an additional set of `SeekableStreamSupervisor` lag metrics, based on time in millis behind the latest offsets of the stream, allowing the Kinesis indexing service extension to have a lag metric. `scheduleReporting ` and `emitLag` were pushed down from `KafkaSupervisor` into `SeekableStreamSupervisor`, and two additional abstract methods, `getPartitionRecordLag` and `getPartitionTimeLag` which collect the two types of lag metrics per partition of the stream.
Kafka's existing lag metrics are still collected in the same format for backwards compatibility.
```
ingest/{supervisor type}/lag
ingest/{supervisor type}/maxLag
ingest/{supervisor type}/avgLag
```
and the new time metrics have a `/time` suffix.
```
ingest/{supervisor type}/lag/time
ingest/{supervisor type}/maxLag/time
ingest/{supervisor type}/avgLag/time
```
Currently `KafkaSupervisor` only emits the former set of metrics, and `KinesisSupervisor`, after this PR, will emit only the latter. It seems possible for `KafkaSupervisor` to emit both sets of metrics, but I will save that for a follow-up PR.
<hr>
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] 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](https://github.com/apache/druid/blob/master/licenses.yaml)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
<hr>
##### Key changed/added classes in this PR
* `SeekableStreamSupervisor`
* `KafkaSupervisor`
* `KinesisSupervisor`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jon-wei commented on a change in pull request #9509: add
kinesis lag metric
Posted by GitBox <gi...@apache.org>.
jon-wei commented on a change in pull request #9509: add kinesis lag metric
URL: https://github.com/apache/druid/pull/9509#discussion_r393408804
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -637,6 +688,25 @@ public void close()
this.closed = true;
}
+ @VisibleForTesting
+ public Map<String, Long> getPartitionTimeLag()
+ {
+ return partitionResources.entrySet()
+ .stream()
+ .collect(
+ Collectors.toMap(k -> k.getKey().getPartitionId(), k -> k.getValue().getPartitionTimeLag())
+ );
+ }
+
+ public Map<String, Long> getPartitionTimeLag(Map<String, String> currentOffsets)
+ {
+ return partitionResources.entrySet()
Review comment:
The number of partitions can be on the order of hundreds, would it make sense to avoid streams here for performance (I remember various issues in the past around stream perf)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jon-wei merged pull request #9509: add kinesis lag metric
Posted by GitBox <gi...@apache.org>.
jon-wei merged pull request #9509: add kinesis lag metric
URL: https://github.com/apache/druid/pull/9509
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jon-wei commented on a change in pull request #9509: add
kinesis lag metric
Posted by GitBox <gi...@apache.org>.
jon-wei commented on a change in pull request #9509: add kinesis lag metric
URL: https://github.com/apache/druid/pull/9509#discussion_r393409891
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
##########
@@ -3158,18 +3167,35 @@ protected abstract void updateLatestSequenceFromStream(
Set<StreamPartition<PartitionIdType>> partitions
);
+ /**
+ * Gets 'lag' of currently processed offset behind latest offset as a measure of difference between offsets.
+ */
+ @Nullable
+ protected abstract Map<PartitionIdType, Long> getPartitionRecordLag();
+
+ /**
+ * Gets 'lag'of currently processed offset behind latest offset as a measure of the difference in time inserted.
Review comment:
'lag'of -> 'lag' of
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jon-wei commented on a change in pull request #9509: add
kinesis lag metric
Posted by GitBox <gi...@apache.org>.
jon-wei commented on a change in pull request #9509: add kinesis lag metric
URL: https://github.com/apache/druid/pull/9509#discussion_r393412399
##########
File path: extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java
##########
@@ -148,6 +149,53 @@ void stopBackgroundFetch()
stopRequested = true;
}
+ long getPartitionTimeLag()
+ {
+ return currentLagMillis;
+ }
+
+ long getPartitionTimeLag(String offset)
+ {
+ // if not started (fetching records in background), fetch lag ourself with a throw-away iterator
+ if (!started) {
+ try {
+ final String iteratorType;
+ final String offsetToUse;
+ if (offset == null || KinesisSupervisor.NOT_SET.equals(offset)) {
+ // this should probably check if will start processing earliest or latest rather than assuming earliest
+ // if latest we could skip this because latest will not be behing latest so lag is 0.
Review comment:
behing -> behind
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] codecov-io commented on issue #9509: add kinesis lag metric
Posted by GitBox <gi...@apache.org>.
codecov-io commented on issue #9509: add kinesis lag metric
URL: https://github.com/apache/druid/pull/9509#issuecomment-598183215
# [Codecov](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=h1) Report
> Merging [#9509](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=desc) into [master](https://codecov.io/gh/apache/druid/commit/7401bb3f9315bf0622b6e0fe7c7fd855d1c65052?src=pr&el=desc) will **increase** coverage by `0.98%`.
> The diff coverage is `57.89%`.
[](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #9509 +/- ##
============================================
+ Coverage 63.56% 64.55% +0.98%
- Complexity 23489 23571 +82
============================================
Files 3040 2932 -108
Lines 125895 119897 -5998
Branches 17410 16048 -1362
============================================
- Hits 80026 77394 -2632
+ Misses 38673 35519 -3154
+ Partials 7196 6984 -212
```
| [Impacted Files](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...pervisor/SeekableStreamSupervisorTuningConfig.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-aW5kZXhpbmctc2VydmljZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHJ1aWQvaW5kZXhpbmcvc2Vla2FibGVzdHJlYW0vc3VwZXJ2aXNvci9TZWVrYWJsZVN0cmVhbVN1cGVydmlzb3JUdW5pbmdDb25maWcuamF2YQ==) | `0% <ø> (ø)` | `0 <0> (ø)` | :arrow_down: |
| [.../kafka/supervisor/KafkaSupervisorTuningConfig.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-ZXh0ZW5zaW9ucy1jb3JlL2thZmthLWluZGV4aW5nLXNlcnZpY2Uvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2RydWlkL2luZGV4aW5nL2thZmthL3N1cGVydmlzb3IvS2Fma2FTdXBlcnZpc29yVHVuaW5nQ29uZmlnLmphdmE=) | `100% <ø> (ø)` | `12 <0> (ø)` | :arrow_down: |
| [...xing/seekablestream/supervisor/TaskReportData.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-aW5kZXhpbmctc2VydmljZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHJ1aWQvaW5kZXhpbmcvc2Vla2FibGVzdHJlYW0vc3VwZXJ2aXNvci9UYXNrUmVwb3J0RGF0YS5qYXZh) | `0% <0%> (ø)` | `0 <0> (ø)` | :arrow_down: |
| [...esis/supervisor/KinesisSupervisorTuningConfig.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-ZXh0ZW5zaW9ucy1jb3JlL2tpbmVzaXMtaW5kZXhpbmctc2VydmljZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHJ1aWQvaW5kZXhpbmcva2luZXNpcy9zdXBlcnZpc29yL0tpbmVzaXNTdXBlcnZpc29yVHVuaW5nQ29uZmlnLmphdmE=) | `100% <100%> (ø)` | `12 <1> (+1)` | :arrow_up: |
| [.../druid/indexing/kinesis/KinesisRecordSupplier.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-ZXh0ZW5zaW9ucy1jb3JlL2tpbmVzaXMtaW5kZXhpbmctc2VydmljZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHJ1aWQvaW5kZXhpbmcva2luZXNpcy9LaW5lc2lzUmVjb3JkU3VwcGxpZXIuamF2YQ==) | `55.27% <100%> (+0.84%)` | `35 <2> (+2)` | :arrow_up: |
| [...uid/indexing/kafka/supervisor/KafkaSupervisor.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-ZXh0ZW5zaW9ucy1jb3JlL2thZmthLWluZGV4aW5nLXNlcnZpY2Uvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2RydWlkL2luZGV4aW5nL2thZmthL3N1cGVydmlzb3IvS2Fma2FTdXBlcnZpc29yLmphdmE=) | `79.51% <20%> (+10.49%)` | `24 <1> (-1)` | :arrow_down: |
| [...blestream/supervisor/SeekableStreamSupervisor.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-aW5kZXhpbmctc2VydmljZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHJ1aWQvaW5kZXhpbmcvc2Vla2FibGVzdHJlYW0vc3VwZXJ2aXNvci9TZWVrYWJsZVN0cmVhbVN1cGVydmlzb3IuamF2YQ==) | `27.25% <59.18%> (+1.5%)` | `62 <5> (+7)` | :arrow_up: |
| [...indexing/kinesis/supervisor/KinesisSupervisor.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-ZXh0ZW5zaW9ucy1jb3JlL2tpbmVzaXMtaW5kZXhpbmctc2VydmljZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZHJ1aWQvaW5kZXhpbmcva2luZXNpcy9zdXBlcnZpc29yL0tpbmVzaXNTdXBlcnZpc29yLmphdmE=) | `75% <83.33%> (+0.21%)` | `33 <2> (+1)` | :arrow_up: |
| [...main/java/org/apache/druid/guice/DruidBinders.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kcnVpZC9ndWljZS9EcnVpZEJpbmRlcnMuamF2YQ==) | `33.33% <0%> (-8.34%)` | `3% <0%> (ø)` | |
| [...org/apache/druid/segment/realtime/FireHydrant.java](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kcnVpZC9zZWdtZW50L3JlYWx0aW1lL0ZpcmVIeWRyYW50LmphdmE=) | `58.33% <0%> (-8.34%)` | `14% <0%> (-1%)` | |
| ... and [148 more](https://codecov.io/gh/apache/druid/pull/9509/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=footer). Last update [7401bb3...646a70b](https://codecov.io/gh/apache/druid/pull/9509?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org