RND-364-async-tasks-improvements

db/db.go

@@ -580,6 +580,8 @@ func createTables(MetadataDbClient MetadataStorage) error {
 			tenant_name VARCHAR NOT NULL DEFAULT '$memphis',
 			station_id INT NOT NULL,
 			created_by VARCHAR NOT NULL,
+			status VARCHAR NOT NULL DEFAULT 'running',
+			failure_reason VARCHAR NOT NULL DEFAULT '',
 			PRIMARY KEY (id)
         );`
 
@@ -603,6 +605,8 @@ func createTables(MetadataDbClient MetadataStorage) error {
 			SELECT 1 FROM information_schema.tables WHERE table_name = 'async_tasks' AND table_schema = 'public'
 		) THEN
 		ALTER TABLE async_tasks ADD COLUMN IF NOT EXISTS created_by VARCHAR NOT NULL;
+		ALTER TABLE async_tasks ADD COLUMN IF NOT EXISTS status VARCHAR NOT NULL DEFAULT 'running';
+		ALTER TABLE async_tasks ADD COLUMN IF NOT EXISTS failure_reason VARCHAR NOT NULL DEFAULT '';
 		END IF;
 		IF EXISTS (
 			SELECT 1 FROM information_schema.table_constraints
@@ -7245,15 +7249,15 @@ func UpsertAsyncTask(task, brokerInCharge string, createdAt time.Time, tenantNam
 	}
 	defer conn.Release()
 
-	query := `INSERT INTO async_tasks (name, broker_in_charge, created_at, updated_at, tenant_name, station_id, created_by) VALUES($1, $2, $3, $4, $5, $6, $7) RETURNING *`
+	query := `INSERT INTO async_tasks (name, broker_in_charge, created_at, updated_at, tenant_name, station_id, created_by, status) VALUES($1, $2, $3, $4, $5, $6, $7, $8) RETURNING *`
 	stmt, err := conn.Conn().Prepare(ctx, "upsert_async_task", query)
 	if err != nil {
 		return models.AsyncTask{}, err
 	}
 
 	updatedAt := createdAt
 	var asyncTask models.AsyncTask
-	err = conn.Conn().QueryRow(ctx, stmt.Name, task, brokerInCharge, createdAt, updatedAt, tenantName, stationId, username).Scan(
+	err = conn.Conn().QueryRow(ctx, stmt.Name, task, brokerInCharge, createdAt, updatedAt, tenantName, stationId, username, "running").Scan(
 		&asyncTask.ID,
 		&asyncTask.Name,
 		&asyncTask.BrokrInCharge,
@@ -7263,6 +7267,8 @@ func UpsertAsyncTask(task, brokerInCharge string, createdAt time.Time, tenantNam
 		&asyncTask.TenantName,
 		&asyncTask.StationId,
 		&asyncTask.CreatedBy,
+		&asyncTask.Status,
+		&asyncTask.FailureReason,
 	)
 
 	if err != nil {
@@ -7370,7 +7376,7 @@ func GetAsyncTaskByNameAndBrokerName(task, brokerName string) (bool, []models.As
 	return true, asyncTask, nil
 }
 
-func GetAllAsyncTasks(tenantName string) ([]models.AsyncTaskRes, error) {
+func GetActiveAndUpdatedAsyncTasks(tenantName string) ([]models.AsyncTaskRes, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 
@@ -7379,12 +7385,40 @@ func GetAllAsyncTasks(tenantName string) ([]models.AsyncTaskRes, error) {
 		return []models.AsyncTaskRes{}, err
 	}
 	defer conn.Release()
-	query := `SELECT a.id, a.name, a.created_at, a.created_by, s.name, a.meta_data
-	FROM async_tasks AS a
-	LEFT JOIN stations AS s ON a.station_id = s.id
-	WHERE a.tenant_name = $1
-	ORDER BY a.created_at DESC;`
-	stmt, err := conn.Conn().Prepare(ctx, "get_all_async_tasks", query)
+	query := `SELECT
+	a.id,
+	a.name,
+	a.created_at,
+	a.created_by,
+	s.name AS station_name,
+	a.meta_data,
+	a.status,
+	a.failure_reason
+	FROM
+	async_tasks AS a
+	LEFT JOIN
+	stations AS s ON a.station_id = s.id
+	WHERE
+	a.tenant_name = $1
+	AND (a.status = 'running' OR a.id IN (
+		SELECT
+			id
+		FROM
+			async_tasks
+		WHERE
+			tenant_name = $1
+			AND status <> 'running'
+		ORDER BY
+			created_at DESC
+		LIMIT 10
+	))
+	ORDER BY
+	CASE
+		WHEN a.status = 'running' THEN 0
+		ELSE 1
+	END,
+	created_at DESC;`
+	stmt, err := conn.Conn().Prepare(ctx, "get_active_and_updated_async_tasks", query)
 	if err != nil {
 		return []models.AsyncTaskRes{}, err
 	}
@@ -7407,6 +7441,8 @@ func GetAllAsyncTasks(tenantName string) ([]models.AsyncTaskRes, error) {
 			&task.CreatedBy,
 			&sName,
 			&task.Data,
+			&task.Status,
+			&task.FailureReason,
 		)
 		if err != nil {
 			return []models.AsyncTaskRes{}, err
@@ -7453,20 +7489,65 @@ func UpdateAsyncTask(task, tenantName string, updatedAt time.Time, metaData inte
 	return nil
 }
 
-func RemoveAsyncTask(task, tenantName string, stationId int) error {
+func UpdateStatusAsyncTask(task, tenantName, status string, stationId int, failureReason, functionName string) error {
+	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
+	defer cancelfunc()
+	conn, err := MetadataDbClient.Client.Acquire(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+	var query string
+	var preparedStmt string
+	if stationId == -1 {
+		query = `UPDATE async_tasks SET status = $1, failure_reason = $2 WHERE name = $3 AND tenant_name=$4 AND station_id = $5 AND meta_data->>'name' = $6`
+		preparedStmt = "edit_status_functions_async_task_by_task_and_tenant_name_and_station_id"
+	} else {
+		query = `UPDATE async_tasks SET status = $1, failure_reason = $2 WHERE name = $3 AND tenant_name=$4 AND station_id = $5`
+		preparedStmt = "edit_status_resend_async_task_by_task_and_tenant_name_and_station_id"
+
+	}
+	stmt, err := conn.Conn().Prepare(ctx, preparedStmt, query)
+	if err != nil {
+		return err
+	}
+	tenantName = strings.ToLower(tenantName)
+	if stationId == -1 {
+		_, err = conn.Conn().Query(ctx, stmt.Name, status, failureReason, task, tenantName, stationId, functionName)
+		if err != nil {
+			return err
+		}
+	} else {
+		_, err = conn.Conn().Query(ctx, stmt.Name, status, failureReason, task, tenantName, stationId)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func RemoveOldAsyncTasks() error {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 	conn, err := MetadataDbClient.Client.Acquire(ctx)
 	if err != nil {
 		return err
 	}
 	defer conn.Release()
-	query := `DELETE FROM async_tasks WHERE name = $1 AND tenant_name=$2 AND station_id = $3`
-	stmt, err := conn.Conn().Prepare(ctx, "remove_async_task_by_name_and_tenant_name_and_station_id", query)
+	query := `DELETE FROM async_tasks
+	WHERE status != 'running' AND id NOT IN (
+		SELECT id
+		FROM async_tasks
+		WHERE status != 'running'
+		ORDER BY updated_at DESC
+		LIMIT 10
+	);`
+	stmt, err := conn.Conn().Prepare(ctx, "remove_old_async_task", query)
 	if err != nil {
 		return err
 	}
-	_, err = conn.Conn().Exec(ctx, stmt.Name, task, tenantName, stationId)
+	_, err = conn.Conn().Exec(ctx, stmt.Name)
 	if err != nil {
 		return err
 	}

models/async_tasks.go

@@ -12,15 +12,19 @@ type AsyncTask struct {
 	TenantName    string      `json:"tenant_name"`
 	StationId     int         `json:"station_id"`
 	CreatedBy     string      `json:"created_by"`
+	Status        string      `json:"status"`
+	FailureReason string      `json:"failure_reason"`
 }
 
 type AsyncTaskRes struct {
-	ID          int         `json:"id"`
-	Name        string      `json:"name"`
-	CreatedAt   time.Time   `json:"created_at"`
-	CreatedBy   string      `json:"created_by"`
-	StationName string      `json:"station_name"`
-	Data        interface{} `json:"data"`
+	ID            int         `json:"id"`
+	Name          string      `json:"name"`
+	CreatedAt     time.Time   `json:"created_at"`
+	CreatedBy     string      `json:"created_by"`
+	StationName   string      `json:"station_name"`
+	Data          interface{} `json:"data"`
+	Status        string      `json:"status"`
+	FailureReason string      `json:"failure_reason"`
 }
 
 type MetaData struct {

server/background_tasks.go

@@ -354,8 +354,9 @@ func (s *Server) StartBackgroundTasks() error {
 	go s.ConsumeFunctionTasks()
 	go s.ScaleFunctionWorkers()
 	go s.ConnectorsDeadPodsRescheduler()
+	go s.removeOldAsyncTasks()
 
-  return nil
+	return nil
 }
 
 func (s *Server) uploadMsgsToTier2Storage() {
@@ -853,3 +854,13 @@ func (s *Server) ReleaseStuckLocks() {
 
 	}
 }
+
+func (s *Server) removeOldAsyncTasks() {
+	ticker := time.NewTicker(15 * time.Minute)
+	for range ticker.C {
+		err := db.RemoveOldAsyncTasks()
+		if err != nil {
+			serv.Errorf("RemoveOldAsyncTasks at db.RemoveOldAsyncTasks : %v", err.Error())
+		}
+	}
+}

server/memphis_handlers_async_tasks.go

@@ -28,7 +28,7 @@ type AsyncTasksHandler struct{}
 func (s *Server) CompleteRelevantStuckAsyncTasks() {
 	exist, asyncTasks, err := db.GetAsyncTaskByNameAndBrokerName("resend_all_dls_msgs", s.opts.ServerName)
 	if err != nil {
-		serv.Errorf("CompleteRelevantStuckAsyncTasks: failed to get async tasks resend_all_dls_msgs: %v", err.Error())
+		s.Errorf("CompleteRelevantStuckAsyncTasks: failed to get async tasks resend_all_dls_msgs: %v", err.Error())
 		return
 	}
 	if !exist {
@@ -38,22 +38,22 @@ func (s *Server) CompleteRelevantStuckAsyncTasks() {
 	for _, asyncTask := range asyncTasks {
 		exist, station, err := db.GetStationById(asyncTask.StationId, asyncTask.TenantName)
 		if err != nil {
-			serv.Errorf("[tenant: %v]CompleteRelevantStuckAsyncTasks at GetStationById: %v", asyncTask.TenantName, err.Error())
+			s.Errorf("[tenant: %v]CompleteRelevantStuckAsyncTasks at GetStationById: %v", asyncTask.TenantName, err.Error())
 			return
 		}
 		if !exist {
 			errMsg := fmt.Sprintf("Station %v does not exist", station.Name)
-			serv.Warnf("[tenant: %v][user: %v]CompleteRelevantStuckAsyncTasks at GetStationById: %v", asyncTask.TenantName, station.CreatedByUsername, errMsg)
+			s.Warnf("[tenant: %v][user: %v]CompleteRelevantStuckAsyncTasks at GetStationById: %v", asyncTask.TenantName, station.CreatedByUsername, errMsg)
 			continue
 		}
 
 		exist, user, err := memphis_cache.GetUser(station.CreatedByUsername, asyncTask.TenantName, false)
 		if err != nil {
-			serv.Errorf("[tenant:%v][user: %v] CompleteRelevantStuckAsyncTasks could not retrive user model from cache or db error: %v", asyncTask.TenantName, user.Username, err)
+			s.Errorf("[tenant:%v][user: %v] CompleteRelevantStuckAsyncTasks could not retrive user model from cache or db error: %v", asyncTask.TenantName, user.Username, err)
 			continue
 		}
 		if !exist {
-			serv.Warnf("[tenant:%v][user: %v] CompleteRelevantStuckAsyncTasks user does not exist", asyncTask.TenantName, user.Username)
+			s.Warnf("[tenant:%v][user: %v] CompleteRelevantStuckAsyncTasks user does not exist", asyncTask.TenantName, user.Username)
 			continue
 		}
 
@@ -103,9 +103,9 @@ func (ash AsyncTasksHandler) GetAsyncTasks(c *gin.Context) {
 }
 
 func (ash AsyncTasksHandler) GetAllAsyncTasks(tenantName string) ([]models.AsyncTaskRes, error) {
-	asyncTasks, err := db.GetAllAsyncTasks(tenantName)
+	asyncTasks, err := db.GetActiveAndUpdatedAsyncTasks(tenantName)
 	if err != nil {
-		serv.Errorf("GetAllAsyncTasks at GetAllAsyncTasks:  %v", err.Error())
+		serv.Errorf("GetAllAsyncTasks at GetActiveAndUpdatedAsyncTasks:  %v", err.Error())
 		return []models.AsyncTaskRes{}, err
 	}
 	for i, task := range asyncTasks {

server/memphis_handlers_monitoring.go

@@ -1385,6 +1385,7 @@ func (mh MonitoringHandler) GetGraphOverview(c *gin.Context) {
 	if err != nil {
 		serv.Errorf("[tenant: %v][user: %v]%v", user.TenantName, user.Username, err.Error())
 		c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+		return
 	}
 
 	c.IndentedJSON(200, res)