OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

allow setting different recording parameters for each path (#2410) (#2457)

Browse Source
This commit is contained in:
Alessandro Ros
2023-10-07 23:48:37 +02:00
committed by GitHub
parent 9a01ab7fd5
commit 8a633d2b79
10 changed files with 197 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+34 -43
View File
@@ -1029,13 +1029,12 @@ There are 3 ways to change the configuration:
### Authentication
Edit `mediamtx.yml` and replace everything inside section `paths` with the following content:
Edit `mediamtx.yml` and set `publishUser` and `publishPass`:
```yml
paths:
all:
publishUser: myuser
publishPass: mypass
pathDefaults:
publishUser: myuser
publishPass: mypass
```
Only publishers that provide both username and password will be able to proceed:
@@ -1047,13 +1046,9 @@ ffmpeg -re -stream_loop -1 -i file.ts -c copy -f rtsp rtsp://myuser:mypass@local
It's possible to setup authentication for readers too:
```yml
paths:
all:
publishUser: myuser
publishPass: mypass
readUser: user
readPass: userpass
pathDefaults:
readUser: user
readPass: userpass
```
If storing plain credentials in the configuration file is a security problem, username and passwords can be stored as sha256-hashed strings; a string must be hashed with sha256 and encoded with base64:
@@ -1065,10 +1060,9 @@ echo -n "userpass" | openssl dgst -binary -sha256 | openssl base64
Then stored with the `sha256:` prefix:
```yml
paths:
all:
readUser: sha256:j1tsRqDEw9xvq/D7/9tMx6Jh/jMhk3UfjwIB2f1zgMo=
readPass: sha256:BdSWkrdV+ZxFBLUQQY7+7uv9RmiSVA8nrPmjGjJtZQQ=
pathDefaults:
readUser: sha256:j1tsRqDEw9xvq/D7/9tMx6Jh/jMhk3UfjwIB2f1zgMo=
readPass: sha256:BdSWkrdV+ZxFBLUQQY7+7uv9RmiSVA8nrPmjGjJtZQQ=
```
**WARNING**: enable encryption or use a VPN to ensure that no one is intercepting the credentials in transit.
@@ -1133,7 +1127,7 @@ To change the format, codec or compression of a stream, use _FFmpeg_ or _GStream
```yml
paths:
all:
compressed:
original:
runOnReady: >
ffmpeg -i rtsp://localhost:$RTSP_PORT/$MTX_PATH
@@ -1147,17 +1141,18 @@ paths:
To save available streams to disk, set the `record` and the `recordPath` parameter in the configuration file:
```yml
# Record streams to disk.
record: yes
# Path of recording segments.
# Extension is added automatically.
# Available variables are %path (path name), %Y %m %d %H %M %S %f (time in strftime format)
recordPath: ./recordings/%path/%Y-%m-%d_%H-%M-%S-%f
pathDefaults:
# Record streams to disk.
record: yes
# Path of recording segments.
# Extension is added automatically.
# Available variables are %path (path name), %Y %m %d %H %M %S %f (time in strftime format)
recordPath: ./recordings/%path/%Y-%m-%d_%H-%M-%S-%f
```
All available recording parameters are listed in the [sample configuration file](/mediamtx.yml).
Be aware that not all tracks can be saved. A compatibility matrix is available at the beginning of the README.
Be aware that not all codecs can be saved with all formats, as described in the compatibility matrix at the beginning of the README.
To upload recordings to a remote location, you can use _MediaMTX_ together with [rclone](https://github.com/rclone/rclone), a command line tool that provides file synchronization capabilities with a huge variety of services (including S3, FTP, SMB, Google Drive):
@@ -1172,17 +1167,14 @@ To upload recordings to a remote location, you can use _MediaMTX_ together with
3. Place `rclone` into the `runOnInit` and `runOnRecordSegmentComplete` hooks:
```yml
record: yes
pathDefaults:
# this is needed to sync segments after a crash.
# replace myconfig with the name of the rclone config.
runOnInit: rclone sync -v ./recordings myconfig:/my-path/recordings
paths:
mypath:
# this is needed to sync segments after a crash.
# replace myconfig with the name of the rclone config.
runOnInit: rclone sync -v ./recordings myconfig:/my-path/recordings
# this is called when a segment has been finalized.
# replace myconfig with the name of the rclone config.
runOnRecordSegmentComplete: rclone sync -v --min-age=1ms ./recordings myconfig:/my-path/recordings
# this is called when a segment has been finalized.
# replace myconfig with the name of the rclone config.
runOnRecordSegmentComplete: rclone sync -v --min-age=1ms ./recordings myconfig:/my-path/recordings
```
If you want to delete local segments after they are uploaded, replace `rclone sync` with `rclone move`.
@@ -1192,13 +1184,12 @@ To upload recordings to a remote location, you can use _MediaMTX_ together with
To forward incoming streams to another server, use _FFmpeg_ inside the `runOnReady` parameter:
```yml
paths:
all:
runOnReady: >
ffmpeg -i rtsp://localhost:$RTSP_PORT/$MTX_PATH
-c copy
-f rtsp rtsp://another-server/another-path
runOnReadyRestart: yes
pathDefaults:
runOnReady: >
ffmpeg -i rtsp://localhost:$RTSP_PORT/$MTX_PATH
-c copy
-f rtsp rtsp://another-server/another-path
runOnReadyRestart: yes
```
### On-demand publishing
@@ -1382,12 +1373,12 @@ paths:
runOnUnread: curl http://my-custom-server/webhook?path=$MTX_PATH&reader_type=$MTX_READER_TYPE&reader_id=$MTX_READER_ID
```
`runOnRecordSegmentComplete` allows to run a command when a record segment is complete:
`runOnRecordSegmentComplete` allows to run a command when a recording segment is complete:
```yml
paths:
mypath:
# Command to run when a record segment is complete.
# Command to run when a recording segment is complete.
# The following environment variables are available:
# * MTX_PATH: path name
# * RTSP_PORT: RTSP server port
internal/conf/conf.go
+23 -16
View File
@@ -163,12 +163,12 @@ type Conf struct {
SRTAddress string `json:"srtAddress"`
// Record
Record bool `json:"record"`
RecordPath string `json:"recordPath"`
RecordFormat string `json:"recordFormat"`
RecordPartDuration StringDuration `json:"recordPartDuration"`
RecordSegmentDuration StringDuration `json:"recordSegmentDuration"`
RecordDeleteAfter StringDuration `json:"recordDeleteAfter"`
Record *bool `json:"record,omitempty"` // deprecated
RecordPath *string `json:"recordPath,omitempty"` // deprecated
RecordFormat *string `json:"recordFormat,omitempty"` // deprecated
RecordPartDuration *StringDuration `json:"recordPartDuration,omitempty"` // deprecated
RecordSegmentDuration *StringDuration `json:"recordSegmentDuration,omitempty"` // deprecated
RecordDeleteAfter *StringDuration `json:"recordDeleteAfter,omitempty"` // deprecated
// Path defaults
PathDefaults Path `json:"pathDefaults"`
@@ -242,13 +242,6 @@ func (conf *Conf) setDefaults() {
conf.SRT = true
conf.SRTAddress = ":8890"
// Record
conf.RecordPath = "./recordings/%path/%Y-%m-%d_%H-%M-%S-%f"
conf.RecordFormat = "fmp4"
conf.RecordPartDuration = 100 * StringDuration(time.Millisecond)
conf.RecordSegmentDuration = 3600 * StringDuration(time.Second)
conf.RecordDeleteAfter = 24 * 3600 * StringDuration(time.Second)
conf.PathDefaults.setDefaults()
}
@@ -414,9 +407,23 @@ func (conf *Conf) Check() error {
}
// Record
if conf.RecordFormat != "fmp4" {
return fmt.Errorf("unsupported record format '%s'", conf.RecordFormat)
if conf.Record != nil {
conf.PathDefaults.Record = *conf.Record
}
if conf.RecordPath != nil {
conf.PathDefaults.RecordPath = *conf.RecordPath
}
if conf.RecordFormat != nil {
conf.PathDefaults.RecordFormat = *conf.RecordFormat
}
if conf.RecordPartDuration != nil {
conf.PathDefaults.RecordPartDuration = *conf.RecordPartDuration
}
if conf.RecordSegmentDuration != nil {
conf.PathDefaults.RecordSegmentDuration = *conf.RecordSegmentDuration
}
if conf.RecordDeleteAfter != nil {
conf.PathDefaults.RecordDeleteAfter = *conf.RecordDeleteAfter
}
conf.Paths = make(map[string]*Path)
internal/conf/conf_test.go
+5 -1
View File
@@ -51,7 +51,11 @@ func TestConfFromFile(t *testing.T) {
Source: "publisher",
SourceOnDemandStartTimeout: 10 * StringDuration(time.Second),
SourceOnDemandCloseAfter: 10 * StringDuration(time.Second),
Record: true,
RecordPath: "./recordings/%path/%Y-%m-%d_%H-%M-%S-%f",
RecordFormat: "fmp4",
RecordPartDuration: 100000000,
RecordSegmentDuration: 3600000000000,
RecordDeleteAfter: 86400000000000,
OverridePublisher: true,
RPICameraWidth: 1920,
RPICameraHeight: 1080,
internal/conf/path.go
+21 -2
View File
@@ -59,7 +59,14 @@ type Path struct {
SourceOnDemandCloseAfter StringDuration `json:"sourceOnDemandCloseAfter"`
MaxReaders int `json:"maxReaders"`
SRTReadPassphrase string `json:"srtReadPassphrase"`
Record bool `json:"record"`
// Record
Record bool `json:"record"`
RecordPath string `json:"recordPath"`
RecordFormat string `json:"recordFormat"`
RecordPartDuration StringDuration `json:"recordPartDuration"`
RecordSegmentDuration StringDuration `json:"recordSegmentDuration"`
RecordDeleteAfter StringDuration `json:"recordDeleteAfter"`
// Authentication
PublishUser Credential `json:"publishUser"`
@@ -139,7 +146,13 @@ func (pconf *Path) setDefaults() {
pconf.Source = "publisher"
pconf.SourceOnDemandStartTimeout = 10 * StringDuration(time.Second)
pconf.SourceOnDemandCloseAfter = 10 * StringDuration(time.Second)
pconf.Record = true
// Record
pconf.RecordPath = "./recordings/%path/%Y-%m-%d_%H-%M-%S-%f"
pconf.RecordFormat = "fmp4"
pconf.RecordPartDuration = 100 * StringDuration(time.Millisecond)
pconf.RecordSegmentDuration = 3600 * StringDuration(time.Second)
pconf.RecordDeleteAfter = 24 * 3600 * StringDuration(time.Second)
// Publisher
pconf.OverridePublisher = true
@@ -386,6 +399,12 @@ func (pconf *Path) check(conf *Conf, name string) error {
}
}
// Record
if pconf.RecordFormat != "fmp4" {
return fmt.Errorf("unsupported record format '%s'", pconf.RecordFormat)
}
// Publisher
if pconf.DisablePublisherOverride != nil {
internal/core/core.go
+37 -15
View File
@@ -8,6 +8,7 @@ import (
"os/signal"
"path/filepath"
"reflect"
"sort"
"strings"
"time"
@@ -33,6 +34,37 @@ var defaultConfPaths = []string{
"/etc/mediamtx/mediamtx.yml",
}
func gatherCleanerEntries(paths map[string]*conf.Path) []record.CleanerEntry {
out := make(map[record.CleanerEntry]struct{})
for _, pa := range paths {
if pa.Record {
entry := record.CleanerEntry{
RecordPath: pa.RecordPath,
RecordDeleteAfter: time.Duration(pa.RecordDeleteAfter),
}
out[entry] = struct{}{}
}
}
out2 := make([]record.CleanerEntry, len(out))
i := 0
for v := range out {
out2[i] = v
i++
}
sort.Slice(out2, func(i, j int) bool {
if out2[i].RecordPath != out2[j].RecordPath {
return out2[i].RecordPath < out2[j].RecordPath
}
return out2[i].RecordDeleteAfter < out2[j].RecordDeleteAfter
})
return out2
}
var cli struct {
Version bool `help:"print version"`
Confpath string `arg:"" default:""`
@@ -259,12 +291,11 @@ func (p *Core) createResources(initial bool) error {
}
}
if p.conf.Record &&
p.conf.RecordDeleteAfter != 0 &&
cleanerEntries := gatherCleanerEntries(p.conf.Paths)
if len(cleanerEntries) != 0 &&
p.recordCleaner == nil {
p.recordCleaner = record.NewCleaner(
p.conf.RecordPath,
time.Duration(p.conf.RecordDeleteAfter),
cleanerEntries,
p,
)
}
@@ -278,10 +309,6 @@ func (p *Core) createResources(initial bool) error {
p.conf.WriteTimeout,
p.conf.WriteQueueSize,
p.conf.UDPMaxPayloadSize,
p.conf.Record,
p.conf.RecordPath,
p.conf.RecordPartDuration,
p.conf.RecordSegmentDuration,
p.conf.Paths,
p.externalCmdPool,
p.metrics,
@@ -539,9 +566,8 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
closeLogger
closeRecorderCleaner := newConf == nil ||
newConf.Record != p.conf.Record ||
newConf.RecordPath != p.conf.RecordPath ||
newConf.RecordDeleteAfter != p.conf.RecordDeleteAfter
!reflect.DeepEqual(gatherCleanerEntries(newConf.Paths), gatherCleanerEntries(p.conf.Paths)) ||
closeLogger
closePathManager := newConf == nil ||
newConf.ExternalAuthenticationURL != p.conf.ExternalAuthenticationURL ||
@@ -551,10 +577,6 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
newConf.UDPMaxPayloadSize != p.conf.UDPMaxPayloadSize ||
newConf.Record != p.conf.Record ||
newConf.RecordPath != p.conf.RecordPath ||
newConf.RecordPartDuration != p.conf.RecordPartDuration ||
newConf.RecordSegmentDuration != p.conf.RecordSegmentDuration ||
closeMetrics ||
closeLogger
if !closePathManager && !reflect.DeepEqual(newConf.Paths, p.conf.Paths) {
internal/core/path.go
+17 -33
View File
@@ -166,22 +166,18 @@ type pathAPIPathsGetReq struct {
}
type path struct {
rtspAddress string
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
writeQueueSize int
udpMaxPayloadSize int
record bool
recordPath string
recordPartDuration conf.StringDuration
recordSegmentDuration conf.StringDuration
confName string
conf *conf.Path
name string
matches []string
wg *sync.WaitGroup
externalCmdPool *externalcmd.Pool
parent pathParent
rtspAddress string
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
writeQueueSize int
udpMaxPayloadSize int
confName string
conf *conf.Path
name string
matches []string
wg *sync.WaitGroup
externalCmdPool *externalcmd.Pool
parent pathParent
ctx context.Context
ctxCancel func()
@@ -226,10 +222,6 @@ func newPath(
writeTimeout conf.StringDuration,
writeQueueSize int,
udpMaxPayloadSize int,
record bool,
recordPath string,
recordPartDuration conf.StringDuration,
recordSegmentDuration conf.StringDuration,
confName string,
cnf *conf.Path,
name string,
@@ -246,10 +238,6 @@ func newPath(
writeTimeout: writeTimeout,
writeQueueSize: writeQueueSize,
udpMaxPayloadSize: udpMaxPayloadSize,
record: record,
recordPath: recordPath,
recordPartDuration: recordPartDuration,
recordSegmentDuration: recordSegmentDuration,
confName: confName,
conf: cnf,
name: name,
@@ -514,7 +502,7 @@ func (pa *path) doReloadConf(newConf *conf.Path) {
go pa.source.(*sourceStatic).reloadConf(newConf)
}
if pa.recordingEnabled() {
if pa.conf.Record {
if pa.stream != nil && pa.recordAgent == nil {
pa.startRecording()
}
@@ -793,10 +781,6 @@ func (pa *path) shouldClose() bool {
len(pa.readerAddRequestsOnHold) == 0
}
func (pa *path) recordingEnabled() bool {
return pa.record && pa.conf.Record
}
func (pa *path) externalCmdEnv() externalcmd.Environment {
_, port, _ := net.SplitHostPort(pa.rtspAddress)
env := externalcmd.Environment{
@@ -897,7 +881,7 @@ func (pa *path) setReady(desc *description.Session, allocateEncoder bool) error
return err
}
if pa.recordingEnabled() {
if pa.conf.Record {
pa.startRecording()
}
@@ -968,9 +952,9 @@ func (pa *path) setNotReady() {
func (pa *path) startRecording() {
pa.recordAgent = record.NewAgent(
pa.writeQueueSize,
pa.recordPath,
time.Duration(pa.recordPartDuration),
time.Duration(pa.recordSegmentDuration),
pa.conf.RecordPath,
time.Duration(pa.conf.RecordPartDuration),
time.Duration(pa.conf.RecordSegmentDuration),
pa.name,
pa.stream,
func(segmentPath string) {
internal/core/path_manager.go
-16
View File
@@ -73,10 +73,6 @@ type pathManager struct {
writeTimeout conf.StringDuration
writeQueueSize int
udpMaxPayloadSize int
record bool
recordPath string
recordPartDuration conf.StringDuration
recordSegmentDuration conf.StringDuration
pathConfs map[string]*conf.Path
externalCmdPool *externalcmd.Pool
metrics *metrics
@@ -111,10 +107,6 @@ func newPathManager(
writeTimeout conf.StringDuration,
writeQueueSize int,
udpMaxPayloadSize int,
record bool,
recordPath string,
recordPartDuration conf.StringDuration,
recordSegmentDuration conf.StringDuration,
pathConfs map[string]*conf.Path,
externalCmdPool *externalcmd.Pool,
metrics *metrics,
@@ -130,10 +122,6 @@ func newPathManager(
writeTimeout: writeTimeout,
writeQueueSize: writeQueueSize,
udpMaxPayloadSize: udpMaxPayloadSize,
record: record,
recordPath: recordPath,
recordPartDuration: recordPartDuration,
recordSegmentDuration: recordSegmentDuration,
pathConfs: pathConfs,
externalCmdPool: externalCmdPool,
metrics: metrics,
@@ -412,10 +400,6 @@ func (pm *pathManager) createPath(
pm.writeTimeout,
pm.writeQueueSize,
pm.udpMaxPayloadSize,
pm.record,
pm.recordPath,
pm.recordPartDuration,
pm.recordSegmentDuration,
pathConfName,
pathConf,
name,
internal/record/cleaner.go
+33 -23
View File
@@ -38,34 +38,35 @@ func commonPath(v string) string {
return common
}
// Cleaner removes expired recordings from disk.
// CleanerEntry is a cleaner entry.
type CleanerEntry struct {
RecordPath string
RecordDeleteAfter time.Duration
}
// Cleaner removes expired recording segments from disk.
type Cleaner struct {
ctx context.Context
ctxCancel func()
path string
deleteAfter time.Duration
parent logger.Writer
ctx context.Context
ctxCancel func()
entries []CleanerEntry
parent logger.Writer
done chan struct{}
}
// NewCleaner allocates a Cleaner.
func NewCleaner(
recordPath string,
deleteAfter time.Duration,
entries []CleanerEntry,
parent logger.Writer,
) *Cleaner {
recordPath += ".mp4"
ctx, ctxCancel := context.WithCancel(context.Background())
c := &Cleaner{
ctx: ctx,
ctxCancel: ctxCancel,
path: recordPath,
deleteAfter: deleteAfter,
parent: parent,
done: make(chan struct{}),
ctx: ctx,
ctxCancel: ctxCancel,
entries: entries,
parent: parent,
done: make(chan struct{}),
}
go c.run()
@@ -88,8 +89,10 @@ func (c *Cleaner) run() {
defer close(c.done)
interval := 30 * 60 * time.Second
if interval > (c.deleteAfter / 2) {
interval = c.deleteAfter / 2
for _, e := range c.entries {
if interval > (e.RecordDeleteAfter / 2) {
interval = e.RecordDeleteAfter / 2
}
}
c.doRun() //nolint:errcheck
@@ -97,7 +100,7 @@ func (c *Cleaner) run() {
for {
select {
case <-time.After(interval):
c.doRun() //nolint:errcheck
c.doRun()
case <-c.ctx.Done():
return
@@ -105,8 +108,15 @@ func (c *Cleaner) run() {
}
}
func (c *Cleaner) doRun() error {
commonPath := commonPath(c.path)
func (c *Cleaner) doRun() {
for _, e := range c.entries {
c.doRunEntry(&e) //nolint:errcheck
}
}
func (c *Cleaner) doRunEntry(e *CleanerEntry) error {
recordPath := e.RecordPath + ".mp4"
commonPath := commonPath(recordPath)
now := timeNow()
filepath.Walk(commonPath, func(path string, info fs.FileInfo, err error) error { //nolint:errcheck
@@ -115,9 +125,9 @@ func (c *Cleaner) doRun() error {
}
if !info.IsDir() {
params := decodeRecordPath(c.path, path)
params := decodeRecordPath(recordPath, path)
if params != nil {
if now.Sub(params.time) > c.deleteAfter {
if now.Sub(params.time) > e.RecordDeleteAfter {
c.Log(logger.Debug, "removing %s", path)
os.Remove(path)
}
internal/record/cleaner_test.go
+4 -2
View File
@@ -30,8 +30,10 @@ func TestCleaner(t *testing.T) {
require.NoError(t, err)
c := NewCleaner(
recordPath,
10*time.Second,
[]CleanerEntry{{
RecordPath: recordPath,
RecordDeleteAfter: 10 * time.Second,
}},
nilLogger{},
)
defer c.Close()
mediamtx.yml
+23 -25
View File
@@ -245,28 +245,6 @@ srt: yes
# Address of the SRT listener.
srtAddress: :8890
###############################################
# Global settings -> Recording
# Record streams to disk.
record: no
# Path of recording segments.
# Extension is added automatically.
# Available variables are %path (path name), %Y %m %d %H %M %S %f (time in strftime format)
recordPath: ./recordings/%path/%Y-%m-%d_%H-%M-%S-%f
# Format of recorded segments.
# Currently the only available format is fmp4 (fragmented MP4).
recordFormat: fmp4
# fMP4 segments are concatenation of small MP4 files (parts), each with this duration.
# When a system failure occurs, the last part gets lost.
# Therefore, the part duration is equal to the RPO (recovery point objective).
recordPartDuration: 100ms
# Minimum duration of each segment.
recordSegmentDuration: 1h
# Delete segments after this timespan.
# Set to 0s to disable automatic deletion.
recordDeleteAfter: 24h
###############################################
# Default path settings
@@ -311,8 +289,28 @@ pathDefaults:
maxReaders: 0
# SRT encryption passphrase require to read from this path
srtReadPassphrase:
# Record streams to disk (if global recording is enabled).
record: yes
###############################################
# Default path settings -> Recording
# Record streams to disk.
record: no
# Path of recording segments.
# Extension is added automatically.
# Available variables are %path (path name), %Y %m %d %H %M %S %f (time in strftime format)
recordPath: ./recordings/%path/%Y-%m-%d_%H-%M-%S-%f
# Format of recorded segments.
# Currently the only available format is fmp4 (fragmented MP4).
recordFormat: fmp4
# fMP4 segments are concatenation of small MP4 files (parts), each with this duration.
# When a system failure occurs, the last part gets lost.
# Therefore, the part duration is equal to the RPO (recovery point objective).
recordPartDuration: 100ms
# Minimum duration of each segment.
recordSegmentDuration: 1h
# Delete segments after this timespan.
# Set to 0s to disable automatic deletion.
recordDeleteAfter: 24h
###############################################
# Default path settings -> Authentication
@@ -520,7 +518,7 @@ pathDefaults:
# Environment variables are the same of runOnRead.
runOnUnread:
# Command to run when a record segment is complete.
# Command to run when a recording segment is complete.
# The following environment variables are available:
# * MTX_PATH: path name
# * RTSP_PORT: RTSP server port