// Copyright 2020-2021 InfluxData, Inc. All rights reserved. // Use of this source code is governed by MIT // license that can be found in the LICENSE file. package api import ( "context" "fmt" "time" "github.com/influxdata/influxdb-client-go/v2/domain" ) // TaskFilter defines filtering options for FindTasks functions. type TaskFilter struct { // Returns task with a specific name Name string // Filter tasks to a specific organization name. OrgName string // Filter tasks to a specific organization ID. OrgID string // Filter tasks to a specific user ID. User string // Filter tasks by a status--"inactive" or "active". Status domain.TaskStatusType // Return tasks after a specified ID. After string // The number of tasks to return. // Default 100, minimum: 1, maximum 500 Limit int } // RunFilter defines filtering options for FindRun* functions. type RunFilter struct { // Return runs after a specified ID. After string // The number of runs to return. // Default 100, minimum 1, maximum 500. Limit int // Filter runs to those scheduled before this time. BeforeTime time.Time // Filter runs to those scheduled after this time. AfterTime time.Time } // TasksAPI provides methods for managing tasks and task runs in an InfluxDB server. type TasksAPI interface { // FindTasks retrieves tasks according to the filter. More fields can be applied. Filter can be nil. FindTasks(ctx context.Context, filter *TaskFilter) ([]domain.Task, error) // GetTask retrieves a refreshed instance of task. GetTask(ctx context.Context, task *domain.Task) (*domain.Task, error) // GetTaskByID retrieves a task found using taskID. GetTaskByID(ctx context.Context, taskID string) (*domain.Task, error) // CreateTask creates a new task according the task object. // It copies OrgId, Name, Description, Flux, Status and Every or Cron properties. Every and Cron are mutually exclusive. // Every has higher priority. CreateTask(ctx context.Context, task *domain.Task) (*domain.Task, error) // CreateTaskWithEvery creates a new task with the name, flux script and every repetition setting, in the org orgID. // Every means duration values. CreateTaskWithEvery(ctx context.Context, name, flux, every, orgID string) (*domain.Task, error) // CreateTaskWithCron creates a new task with the name, flux script and cron repetition setting, in the org orgID // Cron holds cron-like setting, e.g. once an hour at beginning of the hour "0 * * * *". CreateTaskWithCron(ctx context.Context, name, flux, cron, orgID string) (*domain.Task, error) // CreateTaskByFlux creates a new task with complete definition in flux script, in the org orgID CreateTaskByFlux(ctx context.Context, flux, orgID string) (*domain.Task, error) // UpdateTask updates a task. // It copies Description, Flux, Status, Offset and Every or Cron properties. Every and Cron are mutually exclusive. // Every has higher priority. UpdateTask(ctx context.Context, task *domain.Task) (*domain.Task, error) // DeleteTask deletes a task. DeleteTask(ctx context.Context, task *domain.Task) error // DeleteTaskWithID deletes a task with taskID. DeleteTaskWithID(ctx context.Context, taskID string) error // FindMembers retrieves members of a task. FindMembers(ctx context.Context, task *domain.Task) ([]domain.ResourceMember, error) // FindMembersWithID retrieves members of a task with taskID. FindMembersWithID(ctx context.Context, taskID string) ([]domain.ResourceMember, error) // AddMember adds a member to a task. AddMember(ctx context.Context, task *domain.Task, user *domain.User) (*domain.ResourceMember, error) // AddMemberWithID adds a member with id memberID to a task with taskID. AddMemberWithID(ctx context.Context, taskID, memberID string) (*domain.ResourceMember, error) // RemoveMember removes a member from a task. RemoveMember(ctx context.Context, task *domain.Task, user *domain.User) error // RemoveMemberWithID removes a member with id memberID from a task with taskID. RemoveMemberWithID(ctx context.Context, taskID, memberID string) error // FindOwners retrieves owners of a task. FindOwners(ctx context.Context, task *domain.Task) ([]domain.ResourceOwner, error) // FindOwnersWithID retrieves owners of a task with taskID. FindOwnersWithID(ctx context.Context, taskID string) ([]domain.ResourceOwner, error) // AddOwner adds an owner to a task. AddOwner(ctx context.Context, task *domain.Task, user *domain.User) (*domain.ResourceOwner, error) // AddOwnerWithID adds an owner with id memberID to a task with taskID. AddOwnerWithID(ctx context.Context, taskID, memberID string) (*domain.ResourceOwner, error) // RemoveOwner removes an owner from a task. RemoveOwner(ctx context.Context, task *domain.Task, user *domain.User) error // RemoveOwnerWithID removes a member with id memberID from a task with taskID. RemoveOwnerWithID(ctx context.Context, taskID, memberID string) error // FindRuns retrieves a task runs according the filter. More fields can be applied. Filter can be nil. FindRuns(ctx context.Context, task *domain.Task, filter *RunFilter) ([]domain.Run, error) // FindRunsWithID retrieves runs of a task with taskID according the filter. More fields can be applied. Filter can be nil. FindRunsWithID(ctx context.Context, taskID string, filter *RunFilter) ([]domain.Run, error) // GetRun retrieves a refreshed instance if a task run. GetRun(ctx context.Context, run *domain.Run) (*domain.Run, error) // GetRunByID retrieves a specific task run by taskID and runID GetRunByID(ctx context.Context, taskID, runID string) (*domain.Run, error) // FindRunLogs return all log events for a task run. FindRunLogs(ctx context.Context, run *domain.Run) ([]domain.LogEvent, error) // FindRunLogsWithID return all log events for a run with runID of a task with taskID. FindRunLogsWithID(ctx context.Context, taskID, runID string) ([]domain.LogEvent, error) // RunManually manually start a run of the task now, overriding the current schedule. RunManually(ctx context.Context, task *domain.Task) (*domain.Run, error) // RunManuallyWithID manually start a run of a task with taskID now, overriding the current schedule. RunManuallyWithID(ctx context.Context, taskID string) (*domain.Run, error) // RetryRun retry a task run. RetryRun(ctx context.Context, run *domain.Run) (*domain.Run, error) // RetryRunWithID retry a run with runID of a task with taskID. RetryRunWithID(ctx context.Context, taskID, runID string) (*domain.Run, error) // CancelRun cancels a running task. CancelRun(ctx context.Context, run *domain.Run) error // CancelRunWithID cancels a running task. CancelRunWithID(ctx context.Context, taskID, runID string) error // FindLogs retrieves all logs for a task. FindLogs(ctx context.Context, task *domain.Task) ([]domain.LogEvent, error) // FindLogsWithID retrieves all logs for a task with taskID. FindLogsWithID(ctx context.Context, taskID string) ([]domain.LogEvent, error) // FindLabels retrieves labels of a task. FindLabels(ctx context.Context, task *domain.Task) ([]domain.Label, error) // FindLabelsWithID retrieves labels of a task with taskID. FindLabelsWithID(ctx context.Context, taskID string) ([]domain.Label, error) // AddLabel adds a label to a task. AddLabel(ctx context.Context, task *domain.Task, label *domain.Label) (*domain.Label, error) // AddLabelWithID adds a label with id labelID to a task with taskID. AddLabelWithID(ctx context.Context, taskID, labelID string) (*domain.Label, error) // RemoveLabel removes a label from a task. RemoveLabel(ctx context.Context, task *domain.Task, label *domain.Label) error // RemoveLabelWithID removes a label with id labelID from a task with taskID. RemoveLabelWithID(ctx context.Context, taskID, labelID string) error } // tasksAPI implements TasksAPI type tasksAPI struct { apiClient *domain.Client } // NewTasksAPI creates new instance of TasksAPI func NewTasksAPI(apiClient *domain.Client) TasksAPI { return &tasksAPI{ apiClient: apiClient, } } func (t *tasksAPI) FindTasks(ctx context.Context, filter *TaskFilter) ([]domain.Task, error) { params := &domain.GetTasksParams{} if filter != nil { if filter.Name != "" { params.Name = &filter.Name } if filter.User != "" { params.User = &filter.User } if filter.OrgID != "" { params.OrgID = &filter.OrgID } if filter.OrgName != "" { params.Org = &filter.OrgName } if filter.Status != "" { status := domain.GetTasksParamsStatus(filter.Status) params.Status = &status } if filter.Limit > 0 { params.Limit = &filter.Limit } if filter.After != "" { params.After = &filter.After } } response, err := t.apiClient.GetTasks(ctx, params) if err != nil { return nil, err } return *response.Tasks, nil } func (t *tasksAPI) GetTask(ctx context.Context, task *domain.Task) (*domain.Task, error) { return t.GetTaskByID(ctx, task.Id) } func (t *tasksAPI) GetTaskByID(ctx context.Context, taskID string) (*domain.Task, error) { params := &domain.GetTasksIDAllParams{ TaskID: taskID, } return t.apiClient.GetTasksID(ctx, params) } func (t *tasksAPI) createTask(ctx context.Context, taskReq *domain.TaskCreateRequest) (*domain.Task, error) { params := &domain.PostTasksAllParams{ Body: domain.PostTasksJSONRequestBody(*taskReq), } return t.apiClient.PostTasks(ctx, params) } func createTaskReqDetailed(name, flux string, every, cron *string, orgID string) *domain.TaskCreateRequest { repetition := "" if every != nil { repetition = fmt.Sprintf("every: %s", *every) } else if cron != nil { repetition = fmt.Sprintf(`cron: "%s"`, *cron) } fullFlux := fmt.Sprintf(`option task = { name: "%s", %s } %s`, name, repetition, flux) return createTaskReq(fullFlux, orgID) } func createTaskReq(flux string, orgID string) *domain.TaskCreateRequest { status := domain.TaskStatusTypeActive taskReq := &domain.TaskCreateRequest{ Flux: flux, Status: &status, OrgID: &orgID, } return taskReq } func (t *tasksAPI) CreateTask(ctx context.Context, task *domain.Task) (*domain.Task, error) { taskReq := createTaskReqDetailed(task.Name, task.Flux, task.Every, task.Cron, task.OrgID) taskReq.Description = task.Description taskReq.Status = task.Status return t.createTask(ctx, taskReq) } func (t *tasksAPI) CreateTaskWithEvery(ctx context.Context, name, flux, every, orgID string) (*domain.Task, error) { taskReq := createTaskReqDetailed(name, flux, &every, nil, orgID) return t.createTask(ctx, taskReq) } func (t *tasksAPI) CreateTaskWithCron(ctx context.Context, name, flux, cron, orgID string) (*domain.Task, error) { taskReq := createTaskReqDetailed(name, flux, nil, &cron, orgID) return t.createTask(ctx, taskReq) } func (t *tasksAPI) CreateTaskByFlux(ctx context.Context, flux, orgID string) (*domain.Task, error) { taskReq := createTaskReq(flux, orgID) return t.createTask(ctx, taskReq) } func (t *tasksAPI) DeleteTask(ctx context.Context, task *domain.Task) error { return t.DeleteTaskWithID(ctx, task.Id) } func (t *tasksAPI) DeleteTaskWithID(ctx context.Context, taskID string) error { params := &domain.DeleteTasksIDAllParams{ TaskID: taskID, } return t.apiClient.DeleteTasksID(ctx, params) } func (t *tasksAPI) UpdateTask(ctx context.Context, task *domain.Task) (*domain.Task, error) { params := &domain.PatchTasksIDAllParams{ Body: domain.PatchTasksIDJSONRequestBody(domain.TaskUpdateRequest{ Description: task.Description, Flux: &task.Flux, Name: &task.Name, Offset: task.Offset, Status: task.Status, }), TaskID: task.Id, } if task.Every != nil { params.Body.Every = task.Every } else { params.Body.Cron = task.Cron } return t.apiClient.PatchTasksID(ctx, params) } func (t *tasksAPI) FindMembers(ctx context.Context, task *domain.Task) ([]domain.ResourceMember, error) { return t.FindMembersWithID(ctx, task.Id) } func (t *tasksAPI) FindMembersWithID(ctx context.Context, taskID string) ([]domain.ResourceMember, error) { params := &domain.GetTasksIDMembersAllParams{ TaskID: taskID, } response, err := t.apiClient.GetTasksIDMembers(ctx, params) if err != nil { return nil, err } return *response.Users, nil } func (t *tasksAPI) AddMember(ctx context.Context, task *domain.Task, user *domain.User) (*domain.ResourceMember, error) { return t.AddMemberWithID(ctx, task.Id, *user.Id) } func (t *tasksAPI) AddMemberWithID(ctx context.Context, taskID, memberID string) (*domain.ResourceMember, error) { params := &domain.PostTasksIDMembersAllParams{ TaskID: taskID, Body: domain.PostTasksIDMembersJSONRequestBody{Id: memberID}, } return t.apiClient.PostTasksIDMembers(ctx, params) } func (t *tasksAPI) RemoveMember(ctx context.Context, task *domain.Task, user *domain.User) error { return t.RemoveMemberWithID(ctx, task.Id, *user.Id) } func (t *tasksAPI) RemoveMemberWithID(ctx context.Context, taskID, memberID string) error { params := &domain.DeleteTasksIDMembersIDAllParams{ TaskID: taskID, UserID: memberID, } return t.apiClient.DeleteTasksIDMembersID(ctx, params) } func (t *tasksAPI) FindOwners(ctx context.Context, task *domain.Task) ([]domain.ResourceOwner, error) { return t.FindOwnersWithID(ctx, task.Id) } func (t *tasksAPI) FindOwnersWithID(ctx context.Context, taskID string) ([]domain.ResourceOwner, error) { params := &domain.GetTasksIDOwnersAllParams{ TaskID: taskID, } response, err := t.apiClient.GetTasksIDOwners(ctx, params) if err != nil { return nil, err } return *response.Users, nil } func (t *tasksAPI) AddOwner(ctx context.Context, task *domain.Task, user *domain.User) (*domain.ResourceOwner, error) { return t.AddOwnerWithID(ctx, task.Id, *user.Id) } func (t *tasksAPI) AddOwnerWithID(ctx context.Context, taskID, memberID string) (*domain.ResourceOwner, error) { params := &domain.PostTasksIDOwnersAllParams{ Body: domain.PostTasksIDOwnersJSONRequestBody{Id: memberID}, TaskID: taskID, } return t.apiClient.PostTasksIDOwners(ctx, params) } func (t *tasksAPI) RemoveOwner(ctx context.Context, task *domain.Task, user *domain.User) error { return t.RemoveOwnerWithID(ctx, task.Id, *user.Id) } func (t *tasksAPI) RemoveOwnerWithID(ctx context.Context, taskID, memberID string) error { params := &domain.DeleteTasksIDOwnersIDAllParams{ TaskID: taskID, UserID: memberID, } return t.apiClient.DeleteTasksIDOwnersID(ctx, params) } func (t *tasksAPI) FindRuns(ctx context.Context, task *domain.Task, filter *RunFilter) ([]domain.Run, error) { return t.FindRunsWithID(ctx, task.Id, filter) } func (t *tasksAPI) FindRunsWithID(ctx context.Context, taskID string, filter *RunFilter) ([]domain.Run, error) { params := &domain.GetTasksIDRunsAllParams{TaskID: taskID} if filter != nil { if !filter.AfterTime.IsZero() { params.AfterTime = &filter.AfterTime } if !filter.BeforeTime.IsZero() { params.BeforeTime = &filter.BeforeTime } if filter.Limit > 0 { params.Limit = &filter.Limit } if filter.After != "" { params.After = &filter.After } } response, err := t.apiClient.GetTasksIDRuns(ctx, params) if err != nil { return nil, err } return *response.Runs, nil } func (t *tasksAPI) GetRun(ctx context.Context, run *domain.Run) (*domain.Run, error) { return t.GetRunByID(ctx, *run.TaskID, *run.Id) } func (t *tasksAPI) GetRunByID(ctx context.Context, taskID, runID string) (*domain.Run, error) { params := &domain.GetTasksIDRunsIDAllParams{ TaskID: taskID, RunID: runID, } return t.apiClient.GetTasksIDRunsID(ctx, params) } func (t *tasksAPI) FindRunLogs(ctx context.Context, run *domain.Run) ([]domain.LogEvent, error) { return t.FindRunLogsWithID(ctx, *run.TaskID, *run.Id) } func (t *tasksAPI) FindRunLogsWithID(ctx context.Context, taskID, runID string) ([]domain.LogEvent, error) { params := &domain.GetTasksIDRunsIDLogsAllParams{ TaskID: taskID, RunID: runID, } response, err := t.apiClient.GetTasksIDRunsIDLogs(ctx, params) if err != nil { return nil, err } if response.Events == nil { return nil, fmt.Errorf("logs for task '%s' run '%s 'not found", taskID, runID) } return *response.Events, nil } func (t *tasksAPI) RunManually(ctx context.Context, task *domain.Task) (*domain.Run, error) { return t.RunManuallyWithID(ctx, task.Id) } func (t *tasksAPI) RunManuallyWithID(ctx context.Context, taskID string) (*domain.Run, error) { params := &domain.PostTasksIDRunsAllParams{ TaskID: taskID, } return t.apiClient.PostTasksIDRuns(ctx, params) } func (t *tasksAPI) RetryRun(ctx context.Context, run *domain.Run) (*domain.Run, error) { return t.RetryRunWithID(ctx, *run.TaskID, *run.Id) } func (t *tasksAPI) RetryRunWithID(ctx context.Context, taskID, runID string) (*domain.Run, error) { params := &domain.PostTasksIDRunsIDRetryAllParams{ TaskID: taskID, RunID: runID, } return t.apiClient.PostTasksIDRunsIDRetry(ctx, params) } func (t *tasksAPI) CancelRun(ctx context.Context, run *domain.Run) error { return t.CancelRunWithID(ctx, *run.TaskID, *run.Id) } func (t *tasksAPI) CancelRunWithID(ctx context.Context, taskID, runID string) error { params := &domain.DeleteTasksIDRunsIDAllParams{ TaskID: taskID, RunID: runID, } return t.apiClient.DeleteTasksIDRunsID(ctx, params) } func (t *tasksAPI) FindLogs(ctx context.Context, task *domain.Task) ([]domain.LogEvent, error) { return t.FindLogsWithID(ctx, task.Id) } func (t *tasksAPI) FindLogsWithID(ctx context.Context, taskID string) ([]domain.LogEvent, error) { params := &domain.GetTasksIDLogsAllParams{ TaskID: taskID, } response, err := t.apiClient.GetTasksIDLogs(ctx, params) if err != nil { return nil, err } if response.Events == nil { return nil, fmt.Errorf("logs for task '%s' not found", taskID) } return *response.Events, nil } func (t *tasksAPI) FindLabels(ctx context.Context, task *domain.Task) ([]domain.Label, error) { return t.FindLabelsWithID(ctx, task.Id) } func (t *tasksAPI) FindLabelsWithID(ctx context.Context, taskID string) ([]domain.Label, error) { params := &domain.GetTasksIDLabelsAllParams{ TaskID: taskID, } response, err := t.apiClient.GetTasksIDLabels(ctx, params) if err != nil { return nil, err } if response.Labels == nil { return nil, fmt.Errorf("lables for task '%s' not found", taskID) } return *response.Labels, nil } func (t *tasksAPI) AddLabel(ctx context.Context, task *domain.Task, label *domain.Label) (*domain.Label, error) { return t.AddLabelWithID(ctx, task.Id, *label.Id) } func (t *tasksAPI) AddLabelWithID(ctx context.Context, taskID, labelID string) (*domain.Label, error) { params := &domain.PostTasksIDLabelsAllParams{ Body: domain.PostTasksIDLabelsJSONRequestBody{LabelID: &labelID}, TaskID: taskID, } response, err := t.apiClient.PostTasksIDLabels(ctx, params) if err != nil { return nil, err } return response.Label, nil } func (t *tasksAPI) RemoveLabel(ctx context.Context, task *domain.Task, label *domain.Label) error { return t.RemoveLabelWithID(ctx, task.Id, *label.Id) } func (t *tasksAPI) RemoveLabelWithID(ctx context.Context, taskID, labelID string) error { params := &domain.DeleteTasksIDLabelsIDAllParams{ TaskID: taskID, LabelID: labelID, } return t.apiClient.DeleteTasksIDLabelsID(ctx, params) }