You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2020/02/06 23:55:32 UTC
[GitHub] [incubator-pinot] akshayrai opened a new pull request #5056: [TE]
[data-availability] Scheduling of data availability tasks & some clean up
akshayrai opened a new pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056
This PR is part of the effort to enable data availability (SLA) monitoring across metrics/datasets. This particular PR only deals with the scheduling of data availability tasks.
Scheduling Criteria - we schedule a data availability (SLA checker) task in the following cases.
a. Cron based scheduler - Schedule a DA tasks with every detection task.
b. Event based scheduler - Schedule a DA task whenever we perform a fallback & schedule a detection.
List of changes:
a. Scheduling logic - schedules Data availability tasks as per the criteria above
b. Added a counter(ingraph) to track the number of data availability tasks.
c. Created a generic ThirdEyeScheduler interface for all the scheduling components.
d. Renamed `DetectionPipelineScheduler` to `DetectionScheduler` and `DetectionAlertScheduler ` to `SubscriptionScheduler`.
e. Created a Task utility class - removed duplicate code & updated all references.
f: Added Unit test to verify if the data availability tasks are created correctly.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r378052349
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -176,35 +198,43 @@ private void populateDetectionMapAndDatasetConfigMap(
return res;
}
- private long createDetectionTask(DetectionConfigDTO detectionConfig) throws JsonProcessingException {
- long detectionConfigId = detectionConfig.getId();
+ private DetectionPipelineTaskInfo getDetectionPipelineTaskInfo(DetectionConfigDTO detectionConfig, long end) {
// Make sure start time is not out of DETECTION_TASK_MAX_LOOKBACK_WINDOW
- long end = System.currentTimeMillis();
long start = Math.max(detectionConfig.getLastTimestamp(),
end - ThirdEyeUtils.DETECTION_TASK_MAX_LOOKBACK_WINDOW);
- DetectionPipelineTaskInfo taskInfo = new DetectionPipelineTaskInfo(detectionConfigId, start, end);
+
+ return new DetectionPipelineTaskInfo(detectionConfig.getId(), start, end);
+ }
+
+ private long createTask(TaskConstants.TaskType taskType, DetectionConfigDTO detectionConfig, long end)
+ throws JsonProcessingException {
+ DetectionPipelineTaskInfo taskInfo = getDetectionPipelineTaskInfo(detectionConfig, end);
String taskInfoJson = OBJECT_MAPPER.writeValueAsString(taskInfo);
- TaskDTO taskDTO = new TaskDTO();
- taskDTO.setTaskType(TaskConstants.TaskType.DETECTION);
- taskDTO.setJobName(TaskConstants.TaskType.DETECTION.toString() + "_" + detectionConfigId);
- taskDTO.setStatus(TaskConstants.TaskStatus.WAITING);
- taskDTO.setTaskInfo(taskInfoJson);
- long taskId = taskDAO.save(taskDTO);
- LOG.info("Created detection pipeline task {} with taskId {}", taskDTO, taskId);
- taskLastCreateTimeMap.put(detectionConfigId, end);
- return taskId;
+ TaskDTO taskDTO = TaskUtils.buildTask(detectionConfig.getId(), taskInfoJson, taskType);
+ long id = taskDAO.save(taskDTO);
+ LOG.info("Created {} task {} with taskId {}", taskType, taskDTO, id);
+ return id;
+ }
+
+ private long createDetectionTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
+ return createTask(TaskConstants.TaskType.DETECTION, detectionConfig, end);
+ }
+
+ private long createDataAvailabilityTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
Review comment:
I like the term SLA since it is easy to understand.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376560176
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+ detectionIdToWatermarkMap.put(detectionConfig.getId(), endtime);
ThirdeyeMetricsUtil.eventScheduledTaskCounter.inc();
taskCount++;
} else {
LOG.warn("Unable to schedule a task for {}, because it is out of scheduling window.", detectionConfigId);
}
}
+
+ // Note: Fallback SLA & Data availability SLA are independent of each other.
+ // For example, if an event doesn't arrive within 24 hours, do a fallback.
+ // On the other hand, a user can setup an SLA alert if there is no data for 3 days.
if (needFallback(detectionConfig)) {
LOG.info("Scheduling a task for detection {} due to the fallback mechanism.", detectionConfigId);
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+
+ if (DetectionUtils.isDataAvailabilityCheckEnabled(detectionConfig)) {
+ createDataAvailabilityTask(detectionConfig, endtime);
Review comment:
Do you create data availability task elsewhere? Is this the only place to trigger data availability task?
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377942043
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/DetectionScheduler.java
##########
@@ -120,34 +126,41 @@ public void run() {
LOG.error("Error removing job key {}", jobKey);
}
}
-
} catch (SchedulerException e) {
LOG.error("Error while scheduling detection pipeline", e);
}
}
- private boolean jobUpdated(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(key);
- CronTrigger cronTrigger = (CronTrigger) triggers.get(0);
- String cronInSchedule = cronTrigger.getCronExpression();
+ @Override
+ public Set<JobKey> getScheduledJobs() throws SchedulerException {
+ return this.scheduler.getJobKeys(GroupMatcher.jobGroupEquals(TaskConstants.TaskType.DETECTION.toString()));
+ }
- if (!config.getCron().equals(cronInSchedule)) {
- LOG.info("Cron expression for detection pipeline {} has been changed from {} to {}. " + "Restarting schedule",
- config.getId(), cronInSchedule, config.getCron());
- return true;
- }
- return false;
+ @Override
+ public void shutdown() throws SchedulerException {
+ AnomalyUtils.safelyShutdownExecutionService(executorService, this.getClass());
+ scheduler.shutdown();
}
- private void startJob(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- Trigger trigger =
- TriggerBuilder.newTrigger().withSchedule(CronScheduleBuilder.cronSchedule(config.getCron())).build();
- JobDetail job = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
- this.scheduler.scheduleJob(job, trigger);
- LOG.info(String.format("scheduled detection pipeline job %s.", key.getName()));
+ @Override
+ public void startJob(AbstractBean config, JobKey key) throws SchedulerException {
+ Trigger trigger = TriggerBuilder.newTrigger().withSchedule(
+ CronScheduleBuilder.cronSchedule(((DetectionConfigBean) config).getCron())).build();
+ JobDetail detectionJob = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
+
+ this.scheduler.scheduleJob(detectionJob, trigger);
+ LOG.info(String.format("scheduled detection pipeline job %s", detectionJob.getKey().getName()));
+
+ // Data availability alerts will be scheduled only when enabled by the user.
+ if (DetectionUtils.isDataAvailabilityCheckEnabled((DetectionConfigDTO) config)) {
Review comment:
In this PR yes. This PR is only addressing the availability of data for detection jobs. In the next phase, we will have a DataSLACronScheduler which will allow users to define custom SLAs independent of detection.
Updated description to reflect this.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377941903
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -176,35 +198,43 @@ private void populateDetectionMapAndDatasetConfigMap(
return res;
}
- private long createDetectionTask(DetectionConfigDTO detectionConfig) throws JsonProcessingException {
- long detectionConfigId = detectionConfig.getId();
+ private DetectionPipelineTaskInfo getDetectionPipelineTaskInfo(DetectionConfigDTO detectionConfig, long end) {
// Make sure start time is not out of DETECTION_TASK_MAX_LOOKBACK_WINDOW
- long end = System.currentTimeMillis();
long start = Math.max(detectionConfig.getLastTimestamp(),
end - ThirdEyeUtils.DETECTION_TASK_MAX_LOOKBACK_WINDOW);
- DetectionPipelineTaskInfo taskInfo = new DetectionPipelineTaskInfo(detectionConfigId, start, end);
+
+ return new DetectionPipelineTaskInfo(detectionConfig.getId(), start, end);
+ }
+
+ private long createTask(TaskConstants.TaskType taskType, DetectionConfigDTO detectionConfig, long end)
+ throws JsonProcessingException {
+ DetectionPipelineTaskInfo taskInfo = getDetectionPipelineTaskInfo(detectionConfig, end);
String taskInfoJson = OBJECT_MAPPER.writeValueAsString(taskInfo);
- TaskDTO taskDTO = new TaskDTO();
- taskDTO.setTaskType(TaskConstants.TaskType.DETECTION);
- taskDTO.setJobName(TaskConstants.TaskType.DETECTION.toString() + "_" + detectionConfigId);
- taskDTO.setStatus(TaskConstants.TaskStatus.WAITING);
- taskDTO.setTaskInfo(taskInfoJson);
- long taskId = taskDAO.save(taskDTO);
- LOG.info("Created detection pipeline task {} with taskId {}", taskDTO, taskId);
- taskLastCreateTimeMap.put(detectionConfigId, end);
- return taskId;
+ TaskDTO taskDTO = TaskUtils.buildTask(detectionConfig.getId(), taskInfoJson, taskType);
+ long id = taskDAO.save(taskDTO);
+ LOG.info("Created {} task {} with taskId {}", taskType, taskDTO, id);
+ return id;
+ }
+
+ private long createDetectionTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
+ return createTask(TaskConstants.TaskType.DETECTION, detectionConfig, end);
+ }
+
+ private long createDataAvailabilityTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
Review comment:
@vincentchenjl agreed. Let's use data availability term for referring to the event (data available or not) and use Data SLA for the alerts that are setup based on these data availability events.
I will update accordingly.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377942622
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -74,7 +82,7 @@ public DataAvailabilityTaskScheduler(long delayInSec, long fallBackTimeInSec, lo
this.delayInSec = delayInSec;
this.fallBackTimeInSec = fallBackTimeInSec;
this.schedulingWindowInSec = schedulingWindowInSec;
- this.taskLastCreateTimeMap = new HashMap<>();
+ this.detectionIdToWatermarkMap = new HashMap<>();
Review comment:
Done
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377941932
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+ detectionIdToWatermarkMap.put(detectionConfig.getId(), endtime);
ThirdeyeMetricsUtil.eventScheduledTaskCounter.inc();
taskCount++;
} else {
LOG.warn("Unable to schedule a task for {}, because it is out of scheduling window.", detectionConfigId);
}
}
+
+ // Note: Fallback SLA & Data availability SLA are independent of each other.
+ // For example, if an event doesn't arrive within 24 hours, do a fallback.
+ // On the other hand, a user can setup an SLA alert if there is no data for 3 days.
if (needFallback(detectionConfig)) {
LOG.info("Scheduling a task for detection {} due to the fallback mechanism.", detectionConfigId);
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+
+ if (DetectionUtils.isDataAvailabilityCheckEnabled(detectionConfig)) {
+ createDataAvailabilityTask(detectionConfig, endtime);
Review comment:
Discussed offline. Data SLA task will be created in 2 cases.
a. Whenever the DetectionCronScheduler schedules a detection task.
b. Whenever the DataAvailabilityTaskScheduler does a fallback.
Note that we do not need to schedule a Data SLA task when a Data availability event arrives.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] vincentchenjl commented on a change in pull
request #5056: [TE] [data-availability] Scheduling of data availability
tasks & some clean up
Posted by GitBox <gi...@apache.org>.
vincentchenjl commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376643627
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -176,35 +198,43 @@ private void populateDetectionMapAndDatasetConfigMap(
return res;
}
- private long createDetectionTask(DetectionConfigDTO detectionConfig) throws JsonProcessingException {
- long detectionConfigId = detectionConfig.getId();
+ private DetectionPipelineTaskInfo getDetectionPipelineTaskInfo(DetectionConfigDTO detectionConfig, long end) {
// Make sure start time is not out of DETECTION_TASK_MAX_LOOKBACK_WINDOW
- long end = System.currentTimeMillis();
long start = Math.max(detectionConfig.getLastTimestamp(),
end - ThirdEyeUtils.DETECTION_TASK_MAX_LOOKBACK_WINDOW);
- DetectionPipelineTaskInfo taskInfo = new DetectionPipelineTaskInfo(detectionConfigId, start, end);
+
+ return new DetectionPipelineTaskInfo(detectionConfig.getId(), start, end);
+ }
+
+ private long createTask(TaskConstants.TaskType taskType, DetectionConfigDTO detectionConfig, long end)
+ throws JsonProcessingException {
+ DetectionPipelineTaskInfo taskInfo = getDetectionPipelineTaskInfo(detectionConfig, end);
String taskInfoJson = OBJECT_MAPPER.writeValueAsString(taskInfo);
- TaskDTO taskDTO = new TaskDTO();
- taskDTO.setTaskType(TaskConstants.TaskType.DETECTION);
- taskDTO.setJobName(TaskConstants.TaskType.DETECTION.toString() + "_" + detectionConfigId);
- taskDTO.setStatus(TaskConstants.TaskStatus.WAITING);
- taskDTO.setTaskInfo(taskInfoJson);
- long taskId = taskDAO.save(taskDTO);
- LOG.info("Created detection pipeline task {} with taskId {}", taskDTO, taskId);
- taskLastCreateTimeMap.put(detectionConfigId, end);
- return taskId;
+ TaskDTO taskDTO = TaskUtils.buildTask(detectionConfig.getId(), taskInfoJson, taskType);
+ long id = taskDAO.save(taskDTO);
+ LOG.info("Created {} task {} with taskId {}", taskType, taskDTO, id);
+ return id;
+ }
+
+ private long createDetectionTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
+ return createTask(TaskConstants.TaskType.DETECTION, detectionConfig, end);
+ }
+
+ private long createDataAvailabilityTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
Review comment:
My thought here is that we keep `DataAvailabilityTaskScheduler` here, because all these tasks are scheduled based on data availability. I agree that `createDataAvailabilityTask` is still confusing with the `DataAvailabilityTaskScheduler`, so maybe we can change it to `createDataSLACheckTask`, which is more descriptive.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376561246
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/DetectionScheduler.java
##########
@@ -120,34 +126,41 @@ public void run() {
LOG.error("Error removing job key {}", jobKey);
}
}
-
} catch (SchedulerException e) {
LOG.error("Error while scheduling detection pipeline", e);
}
}
- private boolean jobUpdated(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(key);
- CronTrigger cronTrigger = (CronTrigger) triggers.get(0);
- String cronInSchedule = cronTrigger.getCronExpression();
+ @Override
+ public Set<JobKey> getScheduledJobs() throws SchedulerException {
+ return this.scheduler.getJobKeys(GroupMatcher.jobGroupEquals(TaskConstants.TaskType.DETECTION.toString()));
+ }
- if (!config.getCron().equals(cronInSchedule)) {
- LOG.info("Cron expression for detection pipeline {} has been changed from {} to {}. " + "Restarting schedule",
- config.getId(), cronInSchedule, config.getCron());
- return true;
- }
- return false;
+ @Override
+ public void shutdown() throws SchedulerException {
+ AnomalyUtils.safelyShutdownExecutionService(executorService, this.getClass());
+ scheduler.shutdown();
}
- private void startJob(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- Trigger trigger =
- TriggerBuilder.newTrigger().withSchedule(CronScheduleBuilder.cronSchedule(config.getCron())).build();
- JobDetail job = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
- this.scheduler.scheduleJob(job, trigger);
- LOG.info(String.format("scheduled detection pipeline job %s.", key.getName()));
+ @Override
+ public void startJob(AbstractBean config, JobKey key) throws SchedulerException {
+ Trigger trigger = TriggerBuilder.newTrigger().withSchedule(
+ CronScheduleBuilder.cronSchedule(((DetectionConfigBean) config).getCron())).build();
+ JobDetail detectionJob = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
+
+ this.scheduler.scheduleJob(detectionJob, trigger);
+ LOG.info(String.format("scheduled detection pipeline job %s", detectionJob.getKey().getName()));
+
+ // Data availability alerts will be scheduled only when enabled by the user.
+ if (DetectionUtils.isDataAvailabilityCheckEnabled((DetectionConfigDTO) config)) {
Review comment:
Do we only run availability detection together with the normal detection job?
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] vincentchenjl commented on a change in pull
request #5056: [TE] [data-availability] Scheduling of data availability
tasks & some clean up
Posted by GitBox <gi...@apache.org>.
vincentchenjl commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376649514
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/ThirdEyeScheduler.java
##########
@@ -0,0 +1,53 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one
+ * * or more contributor license agreements. See the NOTICE file
+ * * distributed with this work for additional information
+ * * regarding copyright ownership. The ASF 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 org.apache.pinot.thirdeye.scheduler;
+
+import java.util.Set;
+import org.apache.pinot.thirdeye.datalayer.pojo.AbstractBean;
+import org.quartz.JobKey;
+import org.quartz.SchedulerException;
+
+
+/**
+ * Interface for ThirdEye's scheduling components
+ */
+public interface ThirdEyeScheduler extends Runnable {
Review comment:
Should we can call it ThirdEyeCronScheduler to be more specific? I am not sure the interfaces defined here apply to any scheduler that listens to external events, data availability for instance. Even we can put the instance of Quatz scheuler here.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376556012
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -176,35 +198,43 @@ private void populateDetectionMapAndDatasetConfigMap(
return res;
}
- private long createDetectionTask(DetectionConfigDTO detectionConfig) throws JsonProcessingException {
- long detectionConfigId = detectionConfig.getId();
+ private DetectionPipelineTaskInfo getDetectionPipelineTaskInfo(DetectionConfigDTO detectionConfig, long end) {
// Make sure start time is not out of DETECTION_TASK_MAX_LOOKBACK_WINDOW
- long end = System.currentTimeMillis();
long start = Math.max(detectionConfig.getLastTimestamp(),
end - ThirdEyeUtils.DETECTION_TASK_MAX_LOOKBACK_WINDOW);
- DetectionPipelineTaskInfo taskInfo = new DetectionPipelineTaskInfo(detectionConfigId, start, end);
+
+ return new DetectionPipelineTaskInfo(detectionConfig.getId(), start, end);
+ }
+
+ private long createTask(TaskConstants.TaskType taskType, DetectionConfigDTO detectionConfig, long end)
+ throws JsonProcessingException {
+ DetectionPipelineTaskInfo taskInfo = getDetectionPipelineTaskInfo(detectionConfig, end);
String taskInfoJson = OBJECT_MAPPER.writeValueAsString(taskInfo);
- TaskDTO taskDTO = new TaskDTO();
- taskDTO.setTaskType(TaskConstants.TaskType.DETECTION);
- taskDTO.setJobName(TaskConstants.TaskType.DETECTION.toString() + "_" + detectionConfigId);
- taskDTO.setStatus(TaskConstants.TaskStatus.WAITING);
- taskDTO.setTaskInfo(taskInfoJson);
- long taskId = taskDAO.save(taskDTO);
- LOG.info("Created detection pipeline task {} with taskId {}", taskDTO, taskId);
- taskLastCreateTimeMap.put(detectionConfigId, end);
- return taskId;
+ TaskDTO taskDTO = TaskUtils.buildTask(detectionConfig.getId(), taskInfoJson, taskType);
+ long id = taskDAO.save(taskDTO);
+ LOG.info("Created {} task {} with taskId {}", taskType, taskDTO, id);
+ return id;
+ }
+
+ private long createDetectionTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
+ return createTask(TaskConstants.TaskType.DETECTION, detectionConfig, end);
+ }
+
+ private long createDataAvailabilityTask(DetectionConfigDTO detectionConfig, long end) throws JsonProcessingException {
Review comment:
"DataAvailability" is a heavily overloaded term now.
We have DataAvailabilityTaskScheduler which is actually "EventTriggerTaskScheduler".
Here we have another "data availability" task. It is too confusing.
Can you do a refactoring?
Also @vincentchenjl FYI
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] harleyjj commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
harleyjj commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376154771
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/ThirdEyeAnomalyApplication.java
##########
@@ -41,8 +41,8 @@
import org.apache.pinot.thirdeye.datasource.DAORegistry;
import org.apache.pinot.thirdeye.datasource.ThirdEyeCacheRegistry;
import org.apache.pinot.thirdeye.datasource.pinot.resources.PinotDataSourceResource;
-import org.apache.pinot.thirdeye.detection.DetectionPipelineScheduler;
-import org.apache.pinot.thirdeye.detection.alert.DetectionAlertScheduler;
+import org.apache.pinot.thirdeye.scheduler.DetectionScheduler;
+import org.apache.pinot.thirdeye.scheduler.SubscriptionScheduler;
Review comment:
I think these name changes are great for adding clarity and consistency
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai merged pull request #5056: [TE]
[data-availability] Scheduling of data availability tasks & some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai merged pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] jihaozh commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
jihaozh commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376589774
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -74,7 +82,7 @@ public DataAvailabilityTaskScheduler(long delayInSec, long fallBackTimeInSec, lo
this.delayInSec = delayInSec;
this.fallBackTimeInSec = fallBackTimeInSec;
this.schedulingWindowInSec = schedulingWindowInSec;
- this.taskLastCreateTimeMap = new HashMap<>();
+ this.detectionIdToWatermarkMap = new HashMap<>();
Review comment:
Maybe we can rename it to something like `detectionIdToLastTaskCreateTimeMap`? Because I think `watermark` is also overloaded in the context. 🙂
Not clear whether it's the detection last time stamp, task create time or the latest data point.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] vincentchenjl commented on a change in pull
request #5056: [TE] [data-availability] Scheduling of data availability
tasks & some clean up
Posted by GitBox <gi...@apache.org>.
vincentchenjl commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376637576
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -74,7 +82,7 @@ public DataAvailabilityTaskScheduler(long delayInSec, long fallBackTimeInSec, lo
this.delayInSec = delayInSec;
this.fallBackTimeInSec = fallBackTimeInSec;
this.schedulingWindowInSec = schedulingWindowInSec;
- this.taskLastCreateTimeMap = new HashMap<>();
+ this.detectionIdToWatermarkMap = new HashMap<>();
Review comment:
+1
The only case that it stores watermark is when there is no task for the detection at all, which should be minor case.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377960627
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+ detectionIdToWatermarkMap.put(detectionConfig.getId(), endtime);
ThirdeyeMetricsUtil.eventScheduledTaskCounter.inc();
taskCount++;
} else {
LOG.warn("Unable to schedule a task for {}, because it is out of scheduling window.", detectionConfigId);
}
}
+
+ // Note: Fallback SLA & Data availability SLA are independent of each other.
+ // For example, if an event doesn't arrive within 24 hours, do a fallback.
+ // On the other hand, a user can setup an SLA alert if there is no data for 3 days.
if (needFallback(detectionConfig)) {
LOG.info("Scheduling a task for detection {} due to the fallback mechanism.", detectionConfigId);
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+
+ if (DetectionUtils.isDataAvailabilityCheckEnabled(detectionConfig)) {
+ createDataAvailabilityTask(detectionConfig, endtime);
+ LOG.info("Scheduling a task for data availability {} due to the fallback mechanism.", detectionConfigId);
Review comment:
This is a flag to enable/disable data sla checks.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377945526
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
Review comment:
This is just a utility method ensuring both the detection and SLA task have the same end time.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai edited a comment on issue #5056: [TE]
[data-availability] Scheduling of data availability tasks & some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai edited a comment on issue #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#issuecomment-584912753
> Do we need to handle the case that user only want alerts for SLA monitoring without any detection?
@vincentchenjl Yes, we will. But this PR is only addressing the availability of data for detection jobs. In the next phase, we will have a DataSLACronScheduler which will allow users to define custom SLAs independent of detection.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377943234
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/DetectionScheduler.java
##########
@@ -120,34 +126,41 @@ public void run() {
LOG.error("Error removing job key {}", jobKey);
}
}
-
} catch (SchedulerException e) {
LOG.error("Error while scheduling detection pipeline", e);
}
}
- private boolean jobUpdated(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(key);
- CronTrigger cronTrigger = (CronTrigger) triggers.get(0);
- String cronInSchedule = cronTrigger.getCronExpression();
+ @Override
+ public Set<JobKey> getScheduledJobs() throws SchedulerException {
+ return this.scheduler.getJobKeys(GroupMatcher.jobGroupEquals(TaskConstants.TaskType.DETECTION.toString()));
+ }
- if (!config.getCron().equals(cronInSchedule)) {
- LOG.info("Cron expression for detection pipeline {} has been changed from {} to {}. " + "Restarting schedule",
- config.getId(), cronInSchedule, config.getCron());
- return true;
- }
- return false;
+ @Override
+ public void shutdown() throws SchedulerException {
+ AnomalyUtils.safelyShutdownExecutionService(executorService, this.getClass());
+ scheduler.shutdown();
}
- private void startJob(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- Trigger trigger =
- TriggerBuilder.newTrigger().withSchedule(CronScheduleBuilder.cronSchedule(config.getCron())).build();
- JobDetail job = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
- this.scheduler.scheduleJob(job, trigger);
- LOG.info(String.format("scheduled detection pipeline job %s.", key.getName()));
+ @Override
+ public void startJob(AbstractBean config, JobKey key) throws SchedulerException {
+ Trigger trigger = TriggerBuilder.newTrigger().withSchedule(
+ CronScheduleBuilder.cronSchedule(((DetectionConfigBean) config).getCron())).build();
+ JobDetail detectionJob = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
+
+ this.scheduler.scheduleJob(detectionJob, trigger);
+ LOG.info(String.format("scheduled detection pipeline job %s", detectionJob.getKey().getName()));
+
+ // Data availability alerts will be scheduled only when enabled by the user.
+ if (DetectionUtils.isDataAvailabilityCheckEnabled((DetectionConfigDTO) config)) {
+ JobDetail dataAvailabilityJob = JobBuilder.newJob(DetectionDataAvailabilityJob.class).withIdentity(key).build();
+ this.scheduler.scheduleJob(dataAvailabilityJob, trigger);
Review comment:
Yes. This PR is only addressing the availability of data for detection jobs. In the next phase, we will have a DataSLACronScheduler which can have a different cron schedule.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] vincentchenjl commented on a change in pull
request #5056: [TE] [data-availability] Scheduling of data availability
tasks & some clean up
Posted by GitBox <gi...@apache.org>.
vincentchenjl commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376639908
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/ThirdEyeAnomalyApplication.java
##########
@@ -41,8 +41,8 @@
import org.apache.pinot.thirdeye.datasource.DAORegistry;
import org.apache.pinot.thirdeye.datasource.ThirdEyeCacheRegistry;
import org.apache.pinot.thirdeye.datasource.pinot.resources.PinotDataSourceResource;
-import org.apache.pinot.thirdeye.detection.DetectionPipelineScheduler;
-import org.apache.pinot.thirdeye.detection.alert.DetectionAlertScheduler;
+import org.apache.pinot.thirdeye.scheduler.DetectionScheduler;
+import org.apache.pinot.thirdeye.scheduler.SubscriptionScheduler;
Review comment:
`DetectionScheduler` can probably be changed to `DetectionCronScheduler`, since it is not the only scheduler to schedule detection tasks and it is the only cron based scheduler for detection.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] jihaozh commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
jihaozh commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376630723
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/DetectionScheduler.java
##########
@@ -120,34 +126,41 @@ public void run() {
LOG.error("Error removing job key {}", jobKey);
}
}
-
} catch (SchedulerException e) {
LOG.error("Error while scheduling detection pipeline", e);
}
}
- private boolean jobUpdated(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(key);
- CronTrigger cronTrigger = (CronTrigger) triggers.get(0);
- String cronInSchedule = cronTrigger.getCronExpression();
+ @Override
+ public Set<JobKey> getScheduledJobs() throws SchedulerException {
+ return this.scheduler.getJobKeys(GroupMatcher.jobGroupEquals(TaskConstants.TaskType.DETECTION.toString()));
+ }
- if (!config.getCron().equals(cronInSchedule)) {
- LOG.info("Cron expression for detection pipeline {} has been changed from {} to {}. " + "Restarting schedule",
- config.getId(), cronInSchedule, config.getCron());
- return true;
- }
- return false;
+ @Override
+ public void shutdown() throws SchedulerException {
+ AnomalyUtils.safelyShutdownExecutionService(executorService, this.getClass());
+ scheduler.shutdown();
}
- private void startJob(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- Trigger trigger =
- TriggerBuilder.newTrigger().withSchedule(CronScheduleBuilder.cronSchedule(config.getCron())).build();
- JobDetail job = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
- this.scheduler.scheduleJob(job, trigger);
- LOG.info(String.format("scheduled detection pipeline job %s.", key.getName()));
+ @Override
+ public void startJob(AbstractBean config, JobKey key) throws SchedulerException {
+ Trigger trigger = TriggerBuilder.newTrigger().withSchedule(
+ CronScheduleBuilder.cronSchedule(((DetectionConfigBean) config).getCron())).build();
+ JobDetail detectionJob = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
+
+ this.scheduler.scheduleJob(detectionJob, trigger);
+ LOG.info(String.format("scheduled detection pipeline job %s", detectionJob.getKey().getName()));
+
+ // Data availability alerts will be scheduled only when enabled by the user.
+ if (DetectionUtils.isDataAvailabilityCheckEnabled((DetectionConfigDTO) config)) {
+ JobDetail dataAvailabilityJob = JobBuilder.newJob(DetectionDataAvailabilityJob.class).withIdentity(key).build();
+ this.scheduler.scheduleJob(dataAvailabilityJob, trigger);
Review comment:
Does the availability task and the detection task share the same cron?
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376558480
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
Review comment:
Why changing the interface to add endtime?
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r378052610
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
Review comment:
It is an overkill to add one additional parameter just for "ensuring both the detection and SLA task have the same end time".
People look at the util function will probably wonder - "why there is only endtime not starttime".
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on issue #5056: [TE]
[data-availability] Scheduling of data availability tasks & some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on issue #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#issuecomment-584912753
> Do we need to handle the case that user only want alerts for SLA monitoring without any detection?
Yes, we will. But this PR is only addressing the availability of data for detection jobs. In the next phase, we will have a DataSLACronScheduler which will allow users to define custom SLAs independent of detection.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r378052994
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/DetectionScheduler.java
##########
@@ -120,34 +126,41 @@ public void run() {
LOG.error("Error removing job key {}", jobKey);
}
}
-
} catch (SchedulerException e) {
LOG.error("Error while scheduling detection pipeline", e);
}
}
- private boolean jobUpdated(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(key);
- CronTrigger cronTrigger = (CronTrigger) triggers.get(0);
- String cronInSchedule = cronTrigger.getCronExpression();
+ @Override
+ public Set<JobKey> getScheduledJobs() throws SchedulerException {
+ return this.scheduler.getJobKeys(GroupMatcher.jobGroupEquals(TaskConstants.TaskType.DETECTION.toString()));
+ }
- if (!config.getCron().equals(cronInSchedule)) {
- LOG.info("Cron expression for detection pipeline {} has been changed from {} to {}. " + "Restarting schedule",
- config.getId(), cronInSchedule, config.getCron());
- return true;
- }
- return false;
+ @Override
+ public void shutdown() throws SchedulerException {
+ AnomalyUtils.safelyShutdownExecutionService(executorService, this.getClass());
+ scheduler.shutdown();
}
- private void startJob(DetectionConfigDTO config, JobKey key) throws SchedulerException {
- Trigger trigger =
- TriggerBuilder.newTrigger().withSchedule(CronScheduleBuilder.cronSchedule(config.getCron())).build();
- JobDetail job = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
- this.scheduler.scheduleJob(job, trigger);
- LOG.info(String.format("scheduled detection pipeline job %s.", key.getName()));
+ @Override
+ public void startJob(AbstractBean config, JobKey key) throws SchedulerException {
+ Trigger trigger = TriggerBuilder.newTrigger().withSchedule(
+ CronScheduleBuilder.cronSchedule(((DetectionConfigBean) config).getCron())).build();
+ JobDetail detectionJob = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
+
+ this.scheduler.scheduleJob(detectionJob, trigger);
+ LOG.info(String.format("scheduled detection pipeline job %s", detectionJob.getKey().getName()));
+
+ // Data availability alerts will be scheduled only when enabled by the user.
+ if (DetectionUtils.isDataAvailabilityCheckEnabled((DetectionConfigDTO) config)) {
+ JobDetail dataAvailabilityJob = JobBuilder.newJob(DetectionDataAvailabilityJob.class).withIdentity(key).build();
+ this.scheduler.scheduleJob(dataAvailabilityJob, trigger);
Review comment:
DataAvailabilityScheduler should be used for it. Probably we don't need another scheduler.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377945846
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/ThirdEyeScheduler.java
##########
@@ -0,0 +1,53 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one
+ * * or more contributor license agreements. See the NOTICE file
+ * * distributed with this work for additional information
+ * * regarding copyright ownership. The ASF 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 org.apache.pinot.thirdeye.scheduler;
+
+import java.util.Set;
+import org.apache.pinot.thirdeye.datalayer.pojo.AbstractBean;
+import org.quartz.JobKey;
+import org.quartz.SchedulerException;
+
+
+/**
+ * Interface for ThirdEye's scheduling components
+ */
+public interface ThirdEyeScheduler extends Runnable {
Review comment:
Makes sense. Updated it.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] akshayrai commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
akshayrai commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r377941871
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/ThirdEyeAnomalyApplication.java
##########
@@ -41,8 +41,8 @@
import org.apache.pinot.thirdeye.datasource.DAORegistry;
import org.apache.pinot.thirdeye.datasource.ThirdEyeCacheRegistry;
import org.apache.pinot.thirdeye.datasource.pinot.resources.PinotDataSourceResource;
-import org.apache.pinot.thirdeye.detection.DetectionPipelineScheduler;
-import org.apache.pinot.thirdeye.detection.alert.DetectionAlertScheduler;
+import org.apache.pinot.thirdeye.scheduler.DetectionScheduler;
+import org.apache.pinot.thirdeye.scheduler.SubscriptionScheduler;
Review comment:
@xiaohui-sun, yes, I have been consistently trying to enforce the usage of subscription rather than detection_alert in our code (this pr and in the past). detection_alert is a very confusing term.
The database still holds the references as detection_alert which is a little harder to migrate but eventually we should.
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] xiaohui-sun commented on a change in pull request
#5056: [TE] [data-availability] Scheduling of data availability tasks &
some clean up
Posted by GitBox <gi...@apache.org>.
xiaohui-sun commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376550707
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/ThirdEyeAnomalyApplication.java
##########
@@ -41,8 +41,8 @@
import org.apache.pinot.thirdeye.datasource.DAORegistry;
import org.apache.pinot.thirdeye.datasource.ThirdEyeCacheRegistry;
import org.apache.pinot.thirdeye.datasource.pinot.resources.PinotDataSourceResource;
-import org.apache.pinot.thirdeye.detection.DetectionPipelineScheduler;
-import org.apache.pinot.thirdeye.detection.alert.DetectionAlertScheduler;
+import org.apache.pinot.thirdeye.scheduler.DetectionScheduler;
+import org.apache.pinot.thirdeye.scheduler.SubscriptionScheduler;
Review comment:
Do you suggest to change the naming convention?
From DetectionAlert -> Subscription? We have detection_alert_config_index table and all the subscription tasks are with the name "detection_alert_xxx".
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] vincentchenjl commented on a change in pull
request #5056: [TE] [data-availability] Scheduling of data availability
tasks & some clean up
Posted by GitBox <gi...@apache.org>.
vincentchenjl commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376643955
##########
File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/anomaly/detection/trigger/DataAvailabilityTaskScheduler.java
##########
@@ -95,16 +103,30 @@ public void run() {
if (isAllDatasetUpdated(detectionConfig, detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
if (isWithinSchedulingWindow(detection2DatasetMap.get(detectionConfig), datasetConfigMap)) {
//TODO: additional check is required if detection is based on aggregated value across multiple data points
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+ detectionIdToWatermarkMap.put(detectionConfig.getId(), endtime);
ThirdeyeMetricsUtil.eventScheduledTaskCounter.inc();
taskCount++;
} else {
LOG.warn("Unable to schedule a task for {}, because it is out of scheduling window.", detectionConfigId);
}
}
+
+ // Note: Fallback SLA & Data availability SLA are independent of each other.
+ // For example, if an event doesn't arrive within 24 hours, do a fallback.
+ // On the other hand, a user can setup an SLA alert if there is no data for 3 days.
if (needFallback(detectionConfig)) {
LOG.info("Scheduling a task for detection {} due to the fallback mechanism.", detectionConfigId);
- createDetectionTask(detectionConfig);
+ long endtime = System.currentTimeMillis();
+ createDetectionTask(detectionConfig, endtime);
+
+ if (DetectionUtils.isDataAvailabilityCheckEnabled(detectionConfig)) {
+ createDataAvailabilityTask(detectionConfig, endtime);
+ LOG.info("Scheduling a task for data availability {} due to the fallback mechanism.", detectionConfigId);
Review comment:
Why this line is only executed when`DetectionUtils.isDataAvailabilityCheckEnabled(detectionConfig)` is true?
----------------------------------------------------------------
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@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org