-
Notifications
You must be signed in to change notification settings - Fork 3.7k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
More error reporting and stats for ingestion tasks #5418
Changes from 1 commit
278144e
bbda06b
9bb55f5
8867c2f
888fa4d
4589808
d5f1e28
9943b27
60beb1c
0e0e855
c2c132a
8e36c22
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,28 @@ | ||
/* | ||
* Licensed to Metamarkets Group Inc. (Metamarkets) under one | ||
* or more contributor license agreements. See the NOTICE file | ||
* distributed with this work for additional information | ||
* regarding copyright ownership. Metamarkets licenses this file | ||
* to you 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 io.druid.indexer; | ||
|
||
public enum IngestionState | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is this same for all types of tasks? If so, I think it's better to expand TaskState to include these new states because every task is the ingestion task and we don't have to keep two states for them. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I decided to keep them separate, since I mean for IngestionState to be an additional qualifier on the existing states (RUNNING,FAILED,SUCCESS). For example, a task could be RUNNING and in DETERMINE_PARTITIONS, or RUNNING and in BUILD_SEGMENTS, or similarly with FAILED. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sounds good. |
||
{ | ||
NOT_STARTED, | ||
DETERMINE_PARTITIONS, | ||
BUILD_SEGMENTS, | ||
COMPLETED | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,29 @@ | ||
/* | ||
* Licensed to Metamarkets Group Inc. (Metamarkets) under one | ||
* or more contributor license agreements. See the NOTICE file | ||
* distributed with this work for additional information | ||
* regarding copyright ownership. Metamarkets licenses this file | ||
* to you 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 io.druid.indexer; | ||
|
||
import java.util.List; | ||
import java.util.Map; | ||
|
||
public interface TaskMetricsGetter | ||
{ | ||
List<String> getKeys(); | ||
Map<String, Double> getMetrics(); | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,47 @@ | ||
/* | ||
* Licensed to Metamarkets Group Inc. (Metamarkets) under one | ||
* or more contributor license agreements. See the NOTICE file | ||
* distributed with this work for additional information | ||
* regarding copyright ownership. Metamarkets licenses this file | ||
* to you 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 io.druid.indexer; | ||
|
||
import com.google.common.collect.Maps; | ||
|
||
import java.util.Map; | ||
|
||
public class TaskMetricsUtils | ||
{ | ||
public static final String ROWS_PROCESSED = "rowsProcessed"; | ||
public static final String ROWS_PROCESSED_WITH_ERRORS = "rowsProcessedWithErrors"; | ||
public static final String ROWS_UNPARSEABLE = "rowsUnparseable"; | ||
public static final String ROWS_THROWN_AWAY = "rowsThrownAway"; | ||
|
||
public static Map<String, Object> makeIngestionRowMetrics( | ||
long rowsProcessed, | ||
long rowsProcessedWithErrors, | ||
long rowsUnparseable, | ||
long rowsThrownAway | ||
) | ||
{ | ||
Map<String, Object> metricsMap = Maps.newHashMap(); | ||
metricsMap.put(ROWS_PROCESSED, rowsProcessed); | ||
metricsMap.put(ROWS_PROCESSED_WITH_ERRORS, rowsProcessedWithErrors); | ||
metricsMap.put(ROWS_UNPARSEABLE, rowsUnparseable); | ||
metricsMap.put(ROWS_THROWN_AWAY, rowsThrownAway); | ||
return metricsMap; | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -25,6 +25,7 @@ | |
import org.joda.time.DateTime; | ||
|
||
import javax.annotation.Nullable; | ||
import java.util.Map; | ||
import java.util.Objects; | ||
|
||
public class TaskStatusPlus | ||
|
@@ -38,6 +39,15 @@ public class TaskStatusPlus | |
private final TaskLocation location; | ||
private final String dataSource; | ||
|
||
@Nullable | ||
private final Map<String, Object> metrics; | ||
|
||
@Nullable | ||
private final String errorMsg; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I really wanted this! |
||
|
||
@Nullable | ||
private final Map<String, Object> context; | ||
|
||
@JsonCreator | ||
public TaskStatusPlus( | ||
@JsonProperty("id") String id, | ||
|
@@ -47,7 +57,10 @@ public TaskStatusPlus( | |
@JsonProperty("statusCode") @Nullable TaskState state, | ||
@JsonProperty("duration") @Nullable Long duration, | ||
@JsonProperty("location") TaskLocation location, | ||
@JsonProperty("dataSource") String dataSource | ||
@JsonProperty("dataSource") String dataSource, | ||
@JsonProperty("metrics") Map<String, Object> metrics, | ||
@JsonProperty("errorMsg") String errorMsg, | ||
@JsonProperty("context") Map<String, Object> context | ||
) | ||
{ | ||
if (state != null && state.isComplete()) { | ||
|
@@ -61,6 +74,9 @@ public TaskStatusPlus( | |
this.duration = duration; | ||
this.location = Preconditions.checkNotNull(location, "location"); | ||
this.dataSource = dataSource; | ||
this.metrics = metrics; | ||
this.errorMsg = errorMsg; | ||
this.context = context; | ||
} | ||
|
||
@JsonProperty | ||
|
@@ -108,6 +124,27 @@ public TaskLocation getLocation() | |
return location; | ||
} | ||
|
||
@Nullable | ||
@JsonProperty("metrics") | ||
public Map<String, Object> getMetrics() | ||
{ | ||
return metrics; | ||
} | ||
|
||
@Nullable | ||
@JsonProperty("errorMsg") | ||
public String getErrorMsg() | ||
{ | ||
return errorMsg; | ||
} | ||
|
||
@Nullable | ||
@JsonProperty("context") | ||
public Map<String, Object> getContext() | ||
{ | ||
return context; | ||
} | ||
|
||
@Override | ||
public boolean equals(Object o) | ||
{ | ||
|
@@ -138,13 +175,37 @@ public boolean equals(Object o) | |
if (!Objects.equals(duration, that.duration)) { | ||
return false; | ||
} | ||
return location.equals(that.location); | ||
|
||
if (!Objects.equals(location, that.location)) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
return false; | ||
} | ||
|
||
if (!Objects.equals(errorMsg, that.errorMsg)) { | ||
return false; | ||
} | ||
|
||
if (!Objects.equals(location, that.location)) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. dupe There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Fixed |
||
return false; | ||
} | ||
|
||
return Objects.equals(context, that.context); | ||
} | ||
|
||
@Override | ||
public int hashCode() | ||
{ | ||
return Objects.hash(id, type, createdTime, queueInsertionTime, state, duration, location); | ||
return Objects.hash( | ||
id, | ||
type, | ||
createdTime, | ||
queueInsertionTime, | ||
state, | ||
duration, | ||
location, | ||
metrics, | ||
errorMsg, | ||
context | ||
); | ||
} | ||
|
||
@JsonProperty | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,78 @@ | ||
/* | ||
* Licensed to Metamarkets Group Inc. (Metamarkets) under one | ||
* or more contributor license agreements. See the NOTICE file | ||
* distributed with this work for additional information | ||
* regarding copyright ownership. Metamarkets licenses this file | ||
* to you 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 io.druid.utils; | ||
|
||
import com.google.common.base.Preconditions; | ||
|
||
public class CircularBuffer<E> | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Any reason not to use https://google.github.io/guava/releases/23.0/api/docs/com/google/common/collect/EvictingQueue.html? However, it would require a different strategy for There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Decided to keep CircularBuffer for now, since it was already in the codebase and I do want getLatest There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
However, would you add some javadocs to this class? I also think we need some unit tests for this class, but it's not mandatory for this PR. |
||
{ | ||
public E[] getBuffer() | ||
{ | ||
return buffer; | ||
} | ||
|
||
private final E[] buffer; | ||
|
||
private int start = 0; | ||
private int size = 0; | ||
|
||
public CircularBuffer(int capacity) | ||
{ | ||
buffer = (E[]) new Object[capacity]; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Maybe explode with a precondition check that capacity is larger than 0 here instead of exploding out of bounds here There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added a preconditions check |
||
} | ||
|
||
public void add(E item) | ||
{ | ||
buffer[start++] = item; | ||
|
||
if (start >= buffer.length) { | ||
start = 0; | ||
} | ||
|
||
if (size < buffer.length) { | ||
size++; | ||
} | ||
} | ||
|
||
public E getLatest(int index) | ||
{ | ||
int bufferIndex = start - index - 1; | ||
if (bufferIndex < 0) { | ||
bufferIndex = buffer.length + bufferIndex; | ||
} | ||
return buffer[bufferIndex]; | ||
} | ||
|
||
public E get(int index) | ||
{ | ||
Preconditions.checkArgument(index >= 0 && index < size, "invalid index"); | ||
|
||
int bufferIndex = (start - size + index) % buffer.length; | ||
if (bufferIndex < 0) { | ||
bufferIndex += buffer.length; | ||
} | ||
return buffer[bufferIndex]; | ||
} | ||
|
||
public int size() | ||
{ | ||
return size; | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -19,9 +19,24 @@ | |
|
||
package io.druid.indexer; | ||
|
||
import javax.annotation.Nullable; | ||
import java.util.Map; | ||
|
||
/** | ||
*/ | ||
public interface Jobby | ||
{ | ||
boolean run(); | ||
|
||
@Nullable | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Would you please add a javadoc describing when the return value can be null? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added javadoc |
||
default Map<String, Object> getStats() | ||
{ | ||
throw new UnsupportedOperationException("This Jobby does not implement getJobStats()."); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Please add the class name to the exception message. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added class name |
||
} | ||
|
||
@Nullable | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Same here for nullable. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added javadoc |
||
default String getErrorMessage() | ||
{ | ||
throw new UnsupportedOperationException("This Jobby does not implement getErrorMessage()."); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Same here. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added class name |
||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit:
ParseException
supports formatted string.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Used formatted string