Fix not propagating process limits

ffmpeg/ffmpeg.go

@@ -32,6 +32,9 @@ type ProcessConfig struct {
 	Reconnect      bool
 	ReconnectDelay time.Duration
 	StaleTimeout   time.Duration
+	LimitCPU       float64
+	LimitMemory    uint64
+	LimitDuration  time.Duration
 	Command        []string
 	Parser         process.Parser
 	Logger         log.Logger
@@ -117,6 +120,9 @@ func (f *ffmpeg) New(config ProcessConfig) (process.Process, error) {
 		Reconnect:      config.Reconnect,
 		ReconnectDelay: config.ReconnectDelay,
 		StaleTimeout:   config.StaleTimeout,
+		LimitCPU:       config.LimitCPU,
+		LimitMemory:    config.LimitMemory,
+		LimitDuration:  config.LimitDuration,
 		Parser:         config.Parser,
 		Logger:         config.Logger,
 		OnStart:        config.OnStart,

process/process.go

@@ -63,26 +63,19 @@ type Config struct {
 
 // Status represents the current status of a process
 type Status struct {
-	// State is the current state of the process. See stateType for the known states.
+	State    string        // State is the current state of the process. See stateType for the known states.
-	State string
+	States   States        // States is the cumulative history of states the process had.
-
+	Order    string        // Order is the wanted condition of process, either "start" or "stop"
-	// States is the cumulative history of states the process had.
+	Duration time.Duration // Duration is the time since the last change of the state
-	States States
+	Time     time.Time     // Time is the time of the last change of the state
-
+	CPU      struct {
-	// Order is the wanted condition of process, either "start" or "stop"
+		Current float64 // Used CPU in percent
-	Order string
+		Limit   float64 // Limit in percent
-
+	}
-	// Duration is the time since the last change of the state
+	Memory struct {
-	Duration time.Duration
+		Current uint64 // Used memory in bytes
-
+		Limit   uint64 // Limit in bytes
-	// Time is the time of the last change of the state
+	}
-	Time time.Time
-
-	// Used CPU in percent
-	CPU float64
-
-	// Used memory in bytes
-	Memory uint64
 }
 
 // States
@@ -390,6 +383,7 @@ func (p *process) getStateString() string {
 // Status returns the current status of the process
 func (p *process) Status() Status {
 	cpu, memory := p.limits.Current()
+	cpuLimit, memoryLimit := p.limits.Limits()
 
 	p.state.lock.Lock()
 	stateTime := p.state.time
@@ -407,10 +401,14 @@ func (p *process) Status() Status {
 		Order:    order,
 		Duration: time.Since(stateTime),
 		Time:     stateTime,
-		CPU:      cpu,
-		Memory:   memory,
 	}
 
+	s.CPU.Current = cpu
+	s.CPU.Limit = cpuLimit
+
+	s.Memory.Current = memory
+	s.Memory.Limit = memoryLimit
+
 	return s
 }
 

restream/restream.go

@@ -355,6 +355,9 @@ func (r *restream) load() error {
 			Reconnect:      t.config.Reconnect,
 			ReconnectDelay: time.Duration(t.config.ReconnectDelay) * time.Second,
 			StaleTimeout:   time.Duration(t.config.StaleTimeout) * time.Second,
+			LimitCPU:       t.config.LimitCPU,
+			LimitMemory:    t.config.LimitMemory,
+			LimitDuration:  time.Duration(t.config.LimitWaitFor) * time.Second,
 			Command:        t.command,
 			Parser:         t.parser,
 			Logger:         t.logger,
@@ -494,6 +497,9 @@ func (r *restream) createTask(config *app.Config) (*task, error) {
 		Reconnect:      t.config.Reconnect,
 		ReconnectDelay: time.Duration(t.config.ReconnectDelay) * time.Second,
 		StaleTimeout:   time.Duration(t.config.StaleTimeout) * time.Second,
+		LimitCPU:       t.config.LimitCPU,
+		LimitMemory:    t.config.LimitMemory,
+		LimitDuration:  time.Duration(t.config.LimitWaitFor) * time.Second,
 		Command:        t.command,
 		Parser:         t.parser,
 		Logger:         t.logger,
@@ -1179,6 +1185,9 @@ func (r *restream) reloadProcess(id string) error {
 		Reconnect:      t.config.Reconnect,
 		ReconnectDelay: time.Duration(t.config.ReconnectDelay) * time.Second,
 		StaleTimeout:   time.Duration(t.config.StaleTimeout) * time.Second,
+		LimitCPU:       t.config.LimitCPU,
+		LimitMemory:    t.config.LimitMemory,
+		LimitDuration:  time.Duration(t.config.LimitWaitFor) * time.Second,
 		Command:        t.command,
 		Parser:         t.parser,
 		Logger:         t.logger,
@@ -1218,8 +1227,8 @@ func (r *restream) GetProcessState(id string) (*app.State, error) {
 	state.State = status.State
 	state.States.Marshal(status.States)
 	state.Time = status.Time.Unix()
-	state.Memory = status.Memory
+	state.Memory = status.Memory.Current
-	state.CPU = status.CPU
+	state.CPU = status.CPU.Current
 	state.Duration = status.Duration.Round(10 * time.Millisecond).Seconds()
 	state.Reconnect = -1
 	state.Command = make([]string, len(task.command))

restream/restream_test.go

@@ -883,3 +883,26 @@ func TestReplacer(t *testing.T) {
 
 	require.Equal(t, process, rs.tasks["314159265359"].config)
 }
+
+func TestProcessLimit(t *testing.T) {
+	rsi, err := getDummyRestreamer(nil, nil, nil, nil)
+	require.NoError(t, err)
+
+	process := getDummyProcess()
+	process.LimitCPU = 61
+	process.LimitMemory = 42
+	process.Autostart = false
+
+	err = rsi.AddProcess(process)
+	require.NoError(t, err)
+
+	rs := rsi.(*restream)
+
+	task, ok := rs.tasks[process.ID]
+	require.True(t, ok)
+
+	status := task.ffmpeg.Status()
+
+	require.Equal(t, float64(61), status.CPU.Limit)
+	require.Equal(t, uint64(42), status.Memory.Limit)
+}