[MM-56184] Allow passing Kubernetes resource requirements to jobs

config/config.sample.toml

@@ -37,6 +37,11 @@ max_concurrent_jobs = 2
 # The supported units of time are "m" (minutes), "h" (hours) and "d" (days).
 failed_jobs_retention_time = "30d"
 
+# Kubernetes API optionally supports definining resource limits and requests on
+# a per job type basis. Example:
+#[jobs.kubernetes]
+#jobs_resource_requirements = '{"transcribing":{"limits":{"cpu":"4000m"},"requests":{"cpu":"2000m"}},"recording":{"limits":{"cpu":"2000m"},"requests":{"cpu":"1000m"}}}'
+
 [logger]
 # A boolean controlling whether to log to the console.
 enable_console = true

public/job/job.go

@@ -22,6 +22,8 @@ const (
 const (
 	MinSupportedRecorderVersion    = "0.6.0"
 	MinSupportedTranscriberVersion = "0.1.0"
+	RecordingJobPrefix             = "calls-recorder"
+	TranscribingJobPrefix          = "calls-transcriber"
 )
 
 var (

service/config.go

@@ -13,6 +13,8 @@ import (
 	"github.com/mattermost/calls-offloader/logger"
 	"github.com/mattermost/calls-offloader/service/api"
 	"github.com/mattermost/calls-offloader/service/auth"
+	"github.com/mattermost/calls-offloader/service/docker"
+	"github.com/mattermost/calls-offloader/service/kubernetes"
 
 	"github.com/kelseyhightower/envconfig"
 )
@@ -78,10 +80,24 @@ const (
 	JobAPITypeKubernetes            = "kubernetes"
 )
 
+// Alias is needed to implement custom unmarshaler.
+type RetentionTime time.Duration
+
+func (rt *RetentionTime) UnmarshalText(data []byte) error {
+	d, err := parseRetentionTime(string(data))
+	if err != nil {
+		return err
+	}
+	*rt = RetentionTime(d)
+	return nil
+}
+
 type JobsConfig struct {
-	APIType                 JobAPIType    `toml:"api_type"`
-	MaxConcurrentJobs       int           `toml:"max_concurrent_jobs"`
-	FailedJobsRetentionTime time.Duration `toml:"failed_jobs_retention_time" ignored:"true"`
+	APIType                 JobAPIType                  `toml:"api_type"`
+	MaxConcurrentJobs       int                         `toml:"max_concurrent_jobs"`
+	FailedJobsRetentionTime RetentionTime               `toml:"failed_jobs_retention_time" ignored:"true"`
+	Kubernetes              kubernetes.JobServiceConfig `toml:"kubernetes"`
+	Docker                  docker.JobServiceConfig     `toml:"docker"`
 }
 
 // We need some custom parsing since duration doesn't support days.
@@ -110,42 +126,6 @@ func parseRetentionTime(val string) (time.Duration, error) {
 	return d, nil
 }
 
-func (c *JobsConfig) UnmarshalTOML(data interface{}) error {
-	if c == nil {
-		return fmt.Errorf("invalid nil pointer")
-	}
-
-	m, ok := data.(map[string]any)
-	if !ok {
-		return fmt.Errorf("invalid data type")
-	}
-
-	apiType, ok := m["api_type"].(string)
-	if !ok {
-		return fmt.Errorf("invalid api_type type")
-	}
-	c.APIType = JobAPIType(apiType)
-
-	maxConcurrentJobs, ok := m["max_concurrent_jobs"].(int64)
-	if !ok {
-		return fmt.Errorf("invalid max_concurrent_jobs type")
-	}
-	c.MaxConcurrentJobs = int(maxConcurrentJobs)
-
-	if val, ok := m["failed_jobs_retention_time"]; ok {
-		retentionTime, ok := val.(string)
-		if !ok {
-			return fmt.Errorf("invalid failed_jobs_retention_time type")
-		}
-
-		var err error
-		c.FailedJobsRetentionTime, err = parseRetentionTime(retentionTime)
-		return err
-	}
-
-	return nil
-}
-
 func (c JobsConfig) IsValid() error {
 	if c.APIType != JobAPITypeDocker && c.APIType != JobAPITypeKubernetes {
 		return fmt.Errorf("invalid APIType value: %s", c.APIType)
@@ -159,10 +139,17 @@ func (c JobsConfig) IsValid() error {
 		return fmt.Errorf("invalid FailedJobsRetentionTime value: should be a positive duration")
 	}
 
-	if c.FailedJobsRetentionTime > 0 && c.FailedJobsRetentionTime < time.Minute {
+	if c.FailedJobsRetentionTime > 0 && time.Duration(c.FailedJobsRetentionTime) < time.Minute {
 		return fmt.Errorf("invalid FailedJobsRetentionTime value: should be at least one minute")
 	}
 
+	switch c.APIType {
+	case JobAPITypeDocker:
+		return c.Docker.IsValid()
+	case JobAPITypeKubernetes:
+		return c.Kubernetes.IsValid()
+	}
+
 	return nil
 }
 
@@ -179,7 +166,7 @@ func (c *Config) ParseFromEnv() error {
 		if err != nil {
 			return fmt.Errorf("failed to parse FailedJobsRetentionTime: %w", err)
 		}
-		c.Jobs.FailedJobsRetentionTime = d
+		c.Jobs.FailedJobsRetentionTime = RetentionTime(d)
 	}
 
 	return envconfig.Process("", c)

service/config_test.go

@@ -4,10 +4,17 @@
 package service
 
 import (
+	"encoding/json"
 	"os"
 	"testing"
 	"time"
 
+	"github.com/mattermost/calls-offloader/public/job"
+	"github.com/mattermost/calls-offloader/service/kubernetes"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"github.com/stretchr/testify/require"
 )
 
@@ -84,7 +91,7 @@ func TestParseFromEnv(t *testing.T) {
 		var cfg Config
 		err := cfg.ParseFromEnv()
 		require.NoError(t, err)
-		require.Equal(t, time.Hour*24, cfg.Jobs.FailedJobsRetentionTime)
+		require.Equal(t, RetentionTime(time.Hour*24), cfg.Jobs.FailedJobsRetentionTime)
 	})
 
 	t.Run("override", func(t *testing.T) {
@@ -97,4 +104,51 @@ func TestParseFromEnv(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, JobAPITypeDocker, cfg.Jobs.APIType)
 	})
+
+	t.Run("kubernetes.JobsResourceRequirements", func(t *testing.T) {
+		requirements := make(kubernetes.JobsResourceRequirements)
+
+		t.Run("empty", func(t *testing.T) {
+			js, err := json.Marshal(requirements)
+			require.NoError(t, err)
+
+			os.Setenv("JOBS_KUBERNETES_JOBSRESOURCEREQUIREMENTS", string(js))
+			defer os.Unsetenv("JOBS_KUBERNETES_JOBSRESOURCEREQUIREMENTS")
+
+			var cfg Config
+			err = cfg.ParseFromEnv()
+			require.NoError(t, err)
+			require.Equal(t, requirements, cfg.Jobs.Kubernetes.JobsResourceRequirements)
+		})
+
+		t.Run("defined", func(t *testing.T) {
+			requirements[job.TypeRecording] = corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{
+					"cpu": resource.MustParse("1"),
+				},
+				Requests: corev1.ResourceList{
+					"cpu": resource.MustParse("1"),
+				},
+			}
+			requirements[job.TypeTranscribing] = corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{
+					"cpu": resource.MustParse("1"),
+				},
+				Requests: corev1.ResourceList{
+					"cpu": resource.MustParse("1"),
+				},
+			}
+
+			js, err := json.Marshal(requirements)
+			require.NoError(t, err)
+
+			os.Setenv("JOBS_KUBERNETES_JOBSRESOURCEREQUIREMENTS", string(js))
+			defer os.Unsetenv("JOBS_KUBERNETES_JOBSRESOURCEREQUIREMENTS")
+
+			var cfg Config
+			err = cfg.ParseFromEnv()
+			require.NoError(t, err)
+			require.Equal(t, requirements, cfg.Jobs.Kubernetes.JobsResourceRequirements)
+		})
+	})
 }

service/docker/service.go

@@ -34,11 +34,6 @@ const (
 	dockerVolumePath       = "/data"
 )
 
-const (
-	recordingJobPrefix    = "calls-recorder"
-	transcribingJobPrefix = "calls-transcriber"
-)
-
 var (
 	dockerStopTimeout          = 5 * time.Minute
 	dockerRetentionJobInterval = time.Minute
@@ -49,6 +44,18 @@ type JobServiceConfig struct {
 	FailedJobsRetentionTime time.Duration
 }
 
+func (c JobServiceConfig) IsValid() error {
+	if c.MaxConcurrentJobs < 0 {
+		return fmt.Errorf("invalid MaxConcurrentJobs value: should be positive")
+	}
+
+	if c.FailedJobsRetentionTime > 0 && c.FailedJobsRetentionTime < time.Minute {
+		return fmt.Errorf("invalid FailedJobsRetentionTime value: should be at least one minute")
+	}
+
+	return nil
+}
+
 type JobService struct {
 	cfg JobServiceConfig
 	log mlog.LoggerIFace
@@ -261,7 +268,7 @@ func (s *JobService) CreateJob(cfg job.Config, onStopCb job.StopCb) (job.Job, er
 	if err != nil {
 		return job.Job{}, fmt.Errorf("failed to list containers: %w", err)
 	}
-	if len(containers) >= s.cfg.MaxConcurrentJobs {
+	if s.cfg.MaxConcurrentJobs > 0 && len(containers) >= s.cfg.MaxConcurrentJobs {
 		if !devMode {
 			return job.Job{}, fmt.Errorf("max concurrent jobs reached")
 		}
@@ -280,13 +287,13 @@ func (s *JobService) CreateJob(cfg job.Config, onStopCb job.StopCb) (job.Job, er
 		var jobData recorder.RecorderConfig
 		jobData.FromMap(cfg.InputData)
 		jobData.SiteURL = getSiteURLForJob(jobData.SiteURL)
-		jobPrefix = recordingJobPrefix
+		jobPrefix = job.RecordingJobPrefix
 		env = append(env, jobData.ToEnv()...)
 	case job.TypeTranscribing:
 		var jobData transcriber.CallTranscriberConfig
 		jobData.FromMap(cfg.InputData)
 		jobData.SiteURL = getSiteURLForJob(jobData.SiteURL)
-		jobPrefix = transcribingJobPrefix
+		jobPrefix = job.TranscribingJobPrefix
 		env = append(env, jobData.ToEnv()...)
 	}
 

service/helper_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/mattermost/calls-offloader/public"
 	"github.com/mattermost/calls-offloader/service/api"
 	"github.com/mattermost/calls-offloader/service/auth"
+	"github.com/mattermost/calls-offloader/service/docker"
 
 	"github.com/stretchr/testify/require"
 )
@@ -86,6 +87,9 @@ func MakeDefaultCfg(tb testing.TB) *Config {
 		Jobs: JobsConfig{
 			APIType:           JobAPITypeDocker,
 			MaxConcurrentJobs: 2,
+			Docker: docker.JobServiceConfig{
+				MaxConcurrentJobs: 2,
+			},
 		},
 		Logger: logger.Config{
 			EnableConsole: true,

service/jobs_service.go

@@ -6,6 +6,7 @@ package service
 import (
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/mattermost/calls-offloader/public/job"
 	"github.com/mattermost/calls-offloader/service/docker"
@@ -29,15 +30,15 @@ func NewJobService(cfg JobsConfig, log mlog.LoggerIFace) (JobService, error) {
 
 	switch cfg.APIType {
 	case JobAPITypeDocker:
-		return docker.NewJobService(log, docker.JobServiceConfig{
-			MaxConcurrentJobs:       cfg.MaxConcurrentJobs,
-			FailedJobsRetentionTime: cfg.FailedJobsRetentionTime,
-		})
+		cfg.Docker.MaxConcurrentJobs = cfg.MaxConcurrentJobs
+		cfg.Docker.FailedJobsRetentionTime = time.Duration(cfg.FailedJobsRetentionTime)
+		log.Info("creating new job service", mlog.Any("apiType", cfg.APIType), mlog.String("config", fmt.Sprintf("%+v", cfg.Docker)))
+		return docker.NewJobService(log, cfg.Docker)
 	case JobAPITypeKubernetes:
-		return kubernetes.NewJobService(log, kubernetes.JobServiceConfig{
-			MaxConcurrentJobs:       cfg.MaxConcurrentJobs,
-			FailedJobsRetentionTime: cfg.FailedJobsRetentionTime,
-		})
+		cfg.Kubernetes.MaxConcurrentJobs = cfg.MaxConcurrentJobs
+		cfg.Kubernetes.FailedJobsRetentionTime = time.Duration(cfg.FailedJobsRetentionTime)
+		log.Info("creating new job service", mlog.Any("apiType", cfg.APIType), mlog.String("config", fmt.Sprintf("%+v", cfg.Kubernetes)))
+		return kubernetes.NewJobService(log, cfg.Kubernetes)
 	default:
 		return nil, fmt.Errorf("%s API is not implemeneted", cfg.APIType)
 	}