mirror of
https://github.com/hanzoai/team.git
synced 2026-11-01 07:55:31 +00:00
zeekay
8feb24ea1f
The server module named github.com/hanzoai/team/server/v8 — a path that resolves to the retired workspace's repository, not this one — while server/public still published github.com/mattermost/mattermost/server/public. Both are github.com/hanzoteam/server/... now, which is where this code actually lives. That second half is what made the binary unbootable. Upstream's public path stayed claimed by upstream, mattermost-plugin-ai imported it, and the build linked public/model twice; public/plugin's gob.Register then ran twice under one name and panicked at init, before main. Reverting the rename traded the brand for a working binary. Rebranding the plugin alongside it needs no such trade: the agents fork is github.com/hanzoteam/agents/v2 against our public module, so no module named mattermost/mattermost is in the graph at all. Measured: `go list -deps ./cmd/mattermost` names exactly one public/model, and it is ours.
388 lines
13 KiB
Go
388 lines
13 KiB
Go
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
package jobs
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"runtime/pprof"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/hanzoteam/server/server/public/model"
|
|
"github.com/hanzoteam/server/server/public/shared/mlog"
|
|
"github.com/hanzoteam/server/server/public/shared/request"
|
|
"github.com/hanzoteam/server/server/v8/channels/store"
|
|
)
|
|
|
|
const (
|
|
CancelWatcherPollingInterval = 5000
|
|
)
|
|
|
|
// JobLoggerFields returns the logger annotations reflecting the given job metadata.
|
|
func JobLoggerFields(job *model.Job) []mlog.Field {
|
|
if job == nil {
|
|
return nil
|
|
}
|
|
|
|
return []mlog.Field{
|
|
mlog.String("job_id", job.Id),
|
|
mlog.String("job_type", job.Type),
|
|
mlog.Millis("job_create_at", job.CreateAt),
|
|
}
|
|
}
|
|
|
|
func (srv *JobServer) CreateJob(rctx request.CTX, jobType string, jobData map[string]string) (*model.Job, *model.AppError) {
|
|
job, appErr := srv._createJob(rctx, jobType, jobData)
|
|
if appErr != nil {
|
|
return nil, appErr
|
|
}
|
|
|
|
if _, err := srv.Store.Job().Save(job); err != nil {
|
|
return nil, model.NewAppError("CreateJob", "app.job.save.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
return job, nil
|
|
}
|
|
|
|
func (srv *JobServer) CreateJobOnce(rctx request.CTX, jobType string, jobData map[string]string) (*model.Job, *model.AppError) {
|
|
job, appErr := srv._createJob(rctx, jobType, jobData)
|
|
if appErr != nil {
|
|
return nil, appErr
|
|
}
|
|
|
|
if _, err := srv.Store.Job().SaveOnce(job); err != nil {
|
|
return nil, model.NewAppError("CreateJob", "app.job.save.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
return job, nil
|
|
}
|
|
|
|
func (srv *JobServer) CreateJobOnceByTypeAndData(rctx request.CTX, jobType string, jobData map[string]string, data map[string]string) (*model.Job, *model.AppError) {
|
|
job, appErr := srv._createJob(rctx, jobType, jobData)
|
|
if appErr != nil {
|
|
return nil, appErr
|
|
}
|
|
|
|
savedJob, err := srv.Store.Job().SaveOnceByTypeAndData(job, data)
|
|
if err != nil {
|
|
return nil, model.NewAppError("CreateJob", "app.job.save.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
return savedJob, nil
|
|
}
|
|
|
|
func (srv *JobServer) _createJob(rctx request.CTX, jobType string, jobData map[string]string) (*model.Job, *model.AppError) {
|
|
job := model.Job{
|
|
Id: model.NewId(),
|
|
Type: jobType,
|
|
CreateAt: model.GetMillis(),
|
|
Status: model.JobStatusPending,
|
|
Data: jobData,
|
|
}
|
|
|
|
if err := job.IsValid(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if srv.workers.Get(job.Type) == nil {
|
|
return nil, model.NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+job.Id, http.StatusBadRequest)
|
|
}
|
|
|
|
return &job, nil
|
|
}
|
|
|
|
func (srv *JobServer) GetJob(rctx request.CTX, id string) (*model.Job, *model.AppError) {
|
|
job, err := srv.Store.Job().Get(rctx, id)
|
|
if err != nil {
|
|
var nfErr *store.ErrNotFound
|
|
switch {
|
|
case errors.As(err, &nfErr):
|
|
return nil, model.NewAppError("GetJob", "app.job.get.app_error", nil, "", http.StatusNotFound).Wrap(err)
|
|
default:
|
|
return nil, model.NewAppError("GetJob", "app.job.get.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
}
|
|
|
|
return job, nil
|
|
}
|
|
|
|
func (srv *JobServer) ClaimJob(job *model.Job) (*model.Job, *model.AppError) {
|
|
newJob, err := srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JobStatusPending, model.JobStatusInProgress)
|
|
if err != nil {
|
|
return nil, model.NewAppError("ClaimJob", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
if newJob != nil {
|
|
if srv.metrics != nil {
|
|
srv.metrics.IncrementJobActive(newJob.Type)
|
|
}
|
|
srv.publishJobStatus(newJob, model.JobStatusInProgress)
|
|
}
|
|
|
|
return newJob, nil
|
|
}
|
|
|
|
func (srv *JobServer) publishJobStatus(job *model.Job, status string) {
|
|
if srv.publish == nil {
|
|
return
|
|
}
|
|
jobCopy := *job
|
|
jobCopy.Status = status
|
|
jobJSON, err := json.Marshal(&jobCopy)
|
|
if err != nil {
|
|
srv.logger.Warn("Failed to marshal job for WebSocket event", mlog.Err(err))
|
|
return
|
|
}
|
|
message := model.NewWebSocketEvent(model.WebsocketEventJobUpdated, "", "", "", nil, "")
|
|
message.Add("job", string(jobJSON))
|
|
message.GetBroadcast().ContainsSensitiveData = true
|
|
srv.publish(message)
|
|
}
|
|
|
|
func (srv *JobServer) SetJobProgress(job *model.Job, progress int64) *model.AppError {
|
|
job.Status = model.JobStatusInProgress
|
|
job.Progress = progress
|
|
|
|
ret, err := srv.Store.Job().UpdateOptimistically(job, model.JobStatusInProgress)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobProgress", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
if ret != nil {
|
|
srv.publishJobStatus(ret, model.JobStatusInProgress)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (srv *JobServer) SetJobWarning(job *model.Job) *model.AppError {
|
|
ret, err := srv.Store.Job().UpdateStatus(job.Id, model.JobStatusWarning)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobWarning", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
if srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(job.Type)
|
|
}
|
|
|
|
srv.publishJobStatus(ret, model.JobStatusWarning)
|
|
return nil
|
|
}
|
|
|
|
func (srv *JobServer) SetJobSuccess(job *model.Job) *model.AppError {
|
|
ret, err := srv.Store.Job().UpdateStatus(job.Id, model.JobStatusSuccess)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobSuccess", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
if srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(ret.Type)
|
|
}
|
|
|
|
srv.publishJobStatus(ret, model.JobStatusSuccess)
|
|
return nil
|
|
}
|
|
|
|
func (srv *JobServer) SetJobError(job *model.Job, jobError *model.AppError) *model.AppError {
|
|
if jobError == nil {
|
|
ret, err := srv.Store.Job().UpdateStatus(job.Id, model.JobStatusError)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobError", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
if srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(ret.Type)
|
|
}
|
|
|
|
srv.publishJobStatus(ret, model.JobStatusError)
|
|
return nil
|
|
}
|
|
|
|
job.Status = model.JobStatusError
|
|
job.Progress = -1
|
|
if job.Data == nil {
|
|
job.Data = make(map[string]string)
|
|
}
|
|
job.Data["error"] = jobError.Message
|
|
if jobError.DetailedError != "" {
|
|
job.Data["error"] += " — " + jobError.DetailedError
|
|
}
|
|
if wrapped := jobError.Unwrap(); wrapped != nil {
|
|
job.Data["error"] += " — " + wrapped.Error()
|
|
}
|
|
ret, err := srv.Store.Job().UpdateOptimistically(job, model.JobStatusInProgress)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobError", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
if ret != nil && srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(job.Type)
|
|
}
|
|
|
|
if ret == nil {
|
|
ret, err = srv.Store.Job().UpdateOptimistically(job, model.JobStatusCancelRequested)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobError", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
if ret == nil {
|
|
return model.NewAppError("SetJobError", "jobs.set_job_error.update.error", nil, "id="+job.Id, http.StatusInternalServerError)
|
|
}
|
|
}
|
|
|
|
srv.publishJobStatus(ret, ret.Status)
|
|
return nil
|
|
}
|
|
|
|
func (srv *JobServer) SetJobCanceled(job *model.Job) *model.AppError {
|
|
ret, err := srv.Store.Job().UpdateStatus(job.Id, model.JobStatusCanceled)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobCanceled", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
if srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(ret.Type)
|
|
}
|
|
|
|
srv.publishJobStatus(ret, model.JobStatusCanceled)
|
|
return nil
|
|
}
|
|
|
|
func (srv *JobServer) SetJobPending(job *model.Job) *model.AppError {
|
|
ret, err := srv.Store.Job().UpdateStatus(job.Id, model.JobStatusPending)
|
|
if err != nil {
|
|
return model.NewAppError("SetJobPending", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
if srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(ret.Type)
|
|
}
|
|
|
|
srv.publishJobStatus(ret, model.JobStatusPending)
|
|
return nil
|
|
}
|
|
|
|
func (srv *JobServer) UpdateInProgressJobData(job *model.Job) *model.AppError {
|
|
job.Status = model.JobStatusInProgress
|
|
job.LastActivityAt = model.GetMillis()
|
|
if _, err := srv.Store.Job().UpdateOptimistically(job, model.JobStatusInProgress); err != nil {
|
|
return model.NewAppError("UpdateInProgressJobData", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// HandleJobPanic is used to handle panics during the execution of a job. It logs the panic and sets the status for the job.
|
|
// After handling, the method repanics! This method is supposed to be `defer`'d at the start of the job.
|
|
func (srv *JobServer) HandleJobPanic(logger mlog.LoggerIFace, job *model.Job) {
|
|
r := recover()
|
|
if r == nil {
|
|
return
|
|
}
|
|
|
|
sb := &strings.Builder{}
|
|
err := pprof.Lookup("goroutine").WriteTo(sb, 2)
|
|
if err != nil {
|
|
logger.Error("Error fetching goroutine stack", mlog.Err(err))
|
|
}
|
|
logger.Error("Unhandled panic in job", mlog.Any("panic", r), mlog.Any("job", job), mlog.String("stack", sb.String()))
|
|
|
|
rerr, ok := r.(error)
|
|
if !ok {
|
|
rerr = fmt.Errorf("job panic: %v", r)
|
|
}
|
|
|
|
appErr := srv.SetJobError(job, model.NewAppError("HandleJobPanic", "app.job.update.app_error", nil, "", http.StatusInternalServerError))
|
|
if appErr != nil {
|
|
appErr = appErr.Wrap(rerr)
|
|
logger.Error("Failed to set the job status to 'failed'", mlog.Err(appErr), mlog.Any("job", job))
|
|
}
|
|
|
|
panic(r)
|
|
}
|
|
|
|
func (srv *JobServer) RequestCancellation(rctx request.CTX, jobId string) *model.AppError {
|
|
newJob, err := srv.Store.Job().UpdateStatusOptimistically(jobId, model.JobStatusPending, model.JobStatusCanceled)
|
|
if err != nil {
|
|
return model.NewAppError("RequestCancellation", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
if newJob != nil {
|
|
if srv.metrics != nil {
|
|
srv.metrics.DecrementJobActive(newJob.Type)
|
|
}
|
|
srv.publishJobStatus(newJob, model.JobStatusCanceled)
|
|
return nil
|
|
}
|
|
|
|
newJob, err = srv.Store.Job().UpdateStatusOptimistically(jobId, model.JobStatusInProgress, model.JobStatusCancelRequested)
|
|
if err != nil {
|
|
return model.NewAppError("RequestCancellation", "app.job.update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
if newJob != nil {
|
|
srv.publishJobStatus(newJob, model.JobStatusCancelRequested)
|
|
return nil
|
|
}
|
|
|
|
return model.NewAppError("RequestCancellation", "jobs.request_cancellation.status.error", nil, "id="+jobId, http.StatusInternalServerError)
|
|
}
|
|
|
|
func (srv *JobServer) CancellationWatcher(rctx request.CTX, jobId string, cancelChan chan struct{}) {
|
|
for {
|
|
select {
|
|
case <-rctx.Context().Done():
|
|
rctx.Logger().Debug("CancellationWatcher for Job Aborting as job has finished.", mlog.String("job_id", jobId))
|
|
return
|
|
case <-time.After(CancelWatcherPollingInterval * time.Millisecond):
|
|
rctx.Logger().Debug("CancellationWatcher for Job started polling.", mlog.String("job_id", jobId))
|
|
jobStatus, err := srv.Store.Job().Get(rctx, jobId)
|
|
if err != nil {
|
|
rctx.Logger().Warn("Error getting job", mlog.String("job_id", jobId), mlog.Err(err))
|
|
continue
|
|
}
|
|
if jobStatus.Status == model.JobStatusCancelRequested {
|
|
close(cancelChan)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func GenerateNextStartDateTime(now time.Time, nextStartTime time.Time) *time.Time {
|
|
nextTime := time.Date(now.Year(), now.Month(), now.Day(), nextStartTime.Hour(), nextStartTime.Minute(), 0, 0, time.Local)
|
|
|
|
if !now.Before(nextTime) {
|
|
nextTime = nextTime.AddDate(0, 0, 1)
|
|
}
|
|
|
|
return &nextTime
|
|
}
|
|
|
|
func (srv *JobServer) CheckForPendingJobsByType(jobType string) (bool, *model.AppError) {
|
|
count, err := srv.Store.Job().GetCountByStatusAndType(model.JobStatusPending, jobType)
|
|
if err != nil {
|
|
return false, model.NewAppError("CheckForPendingJobsByType", "app.job.get_count_by_status_and_type.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
return count > 0, nil
|
|
}
|
|
|
|
func (srv *JobServer) GetJobsByTypeAndStatus(rctx request.CTX, jobType string, status string) ([]*model.Job, *model.AppError) {
|
|
jobs, err := srv.Store.Job().GetAllByTypeAndStatus(rctx, jobType, status)
|
|
if err != nil {
|
|
return nil, model.NewAppError("GetJobsByTypeAndStatus", "app.job.get_all_jobs_by_type_and_status.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
|
|
return jobs, nil
|
|
}
|
|
|
|
func (srv *JobServer) GetLastSuccessfulJobByType(jobType string) (*model.Job, *model.AppError) {
|
|
statuses := []string{model.JobStatusSuccess}
|
|
if jobType == model.JobTypeMessageExport {
|
|
statuses = []string{model.JobStatusWarning, model.JobStatusSuccess}
|
|
}
|
|
job, err := srv.Store.Job().GetNewestJobByStatusesAndType(statuses, jobType)
|
|
var nfErr *store.ErrNotFound
|
|
if err != nil && !errors.As(err, &nfErr) {
|
|
return nil, model.NewAppError("GetLastSuccessfulJobByType", "app.job.get_newest_job_by_status_and_type.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
|
|
}
|
|
return job, nil
|
|
}
|