mirror of
https://github.com/kerberos-io/agent.git
synced 2026-08-23 15:08:32 +00:00
Merge pull request #286 from kerberos-io/feature/resumable-uploads-tusd
feature/resumable-uploads-tusd
This commit is contained in:
Cédric Verstraeten
2
.gitignore
vendored
2
.gitignore
vendored
@@ -14,5 +14,7 @@ machinery/test*
|
||||
machinery/init-dev.sh
|
||||
machinery/.env.local
|
||||
machinery/vendor
|
||||
machinery/go.work
|
||||
machinery/go.work.sum
|
||||
deployments/docker/private-docker-compose.yaml
|
||||
video.mp4
|
||||
3
.vscode/launch.json
vendored
3
.vscode/launch.json
vendored
@@ -18,6 +18,9 @@
|
||||
],
|
||||
"envFile": "${workspaceFolder}/machinery/.env.local",
|
||||
"buildFlags": "--tags dynamic",
|
||||
"env": {
|
||||
"GOWORK": "off"
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "Launch React",
|
||||
|
||||
@@ -27,5 +27,12 @@ AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY=
|
||||
AGENT_KERBEROSVAULT_SECONDARY_ACCESS_KEY=
|
||||
AGENT_KERBEROSVAULT_SECONDARY_SECRET_KEY=
|
||||
|
||||
# Resumable (tus) uploads to Kerberos Vault are enabled by default.
|
||||
# Set to true to fall back to the legacy single-shot POST /storage upload.
|
||||
#AGENT_DISABLE_RESUMABLE_UPLOAD=true
|
||||
# Bytes sent per PATCH request (default 1 MiB = 1048576). 0 disables chunking
|
||||
# and sends the whole file in a single PATCH.
|
||||
AGENT_TUS_CHUNK_SIZE_BYTES=1048576
|
||||
|
||||
# Open telemetry tracing endpoint
|
||||
OTEL_EXPORTER_OTLP_ENDPOINT=
|
||||
425
machinery/cmd/mp4analyze/main.go
Normal file
425
machinery/cmd/mp4analyze/main.go
Normal file
@@ -0,0 +1,425 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/Eyevinn/mp4ff/avc"
|
||||
mp4ff "github.com/Eyevinn/mp4ff/mp4"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 {
|
||||
fmt.Println("usage: mp4analyze <file.mp4>")
|
||||
os.Exit(1)
|
||||
}
|
||||
f, err := os.Open(os.Args[1])
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
defer f.Close()
|
||||
parsed, err := mp4ff.DecodeFile(f)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
// Movie-level info
|
||||
if parsed.Init != nil && parsed.Init.Moov != nil {
|
||||
moov := parsed.Init.Moov
|
||||
fmt.Printf("ftyp/moov present. timescale(mvhd)=%d duration(mvhd)=%d\n",
|
||||
moov.Mvhd.Timescale, moov.Mvhd.Duration)
|
||||
for _, trak := range moov.Traks {
|
||||
ts := trak.Mdia.Mdhd.Timescale
|
||||
fmt.Printf(" trak id=%d handler=%s mdhd.timescale=%d mdhd.duration=%d\n",
|
||||
trak.Tkhd.TrackID, trak.Mdia.Hdlr.HandlerType, ts, trak.Mdia.Mdhd.Duration)
|
||||
}
|
||||
} else {
|
||||
fmt.Println("no Init/Moov (pure fragmented stream?)")
|
||||
}
|
||||
|
||||
// sidx vs actual segment layout. MSE players use sidx to map presentation
|
||||
// time -> byte ranges; if sidx references disagree with the real segment
|
||||
// sizes/durations (e.g. after an early/short flush) the player fetches the
|
||||
// wrong bytes and fails to decode — a failure that "heals" on seek.
|
||||
fmt.Println("=== sidx references vs actual segments ===")
|
||||
var sidxRefs []mp4ff.SidxRef
|
||||
for _, c := range parsed.Children {
|
||||
if s, ok := c.(*mp4ff.SidxBox); ok {
|
||||
fmt.Printf(" sidx: timescale=%d earliestPresTime=%d firstOffset=%d refCount=%d anchor(after sidx)=%d\n",
|
||||
s.Timescale, s.EarliestPresentationTime, s.FirstOffset, len(s.SidxRefs), s.AnchorPoint)
|
||||
sidxRefs = s.SidxRefs
|
||||
}
|
||||
}
|
||||
// Actual segment sizes (styp+moof+mdat) and fragment durations.
|
||||
type segInfo struct {
|
||||
size uint64
|
||||
dur uint64
|
||||
}
|
||||
var actual []segInfo
|
||||
for _, seg := range parsed.Segments {
|
||||
var sz uint64
|
||||
if seg.Styp != nil {
|
||||
sz += seg.Styp.Size()
|
||||
}
|
||||
if seg.Sidx != nil {
|
||||
sz += seg.Sidx.Size()
|
||||
}
|
||||
var dur uint64
|
||||
for _, fr := range seg.Fragments {
|
||||
sz += fr.Moof.Size()
|
||||
if fr.Mdat != nil {
|
||||
sz += fr.Mdat.Size()
|
||||
}
|
||||
for _, traf := range fr.Moof.Trafs {
|
||||
if traf.Tfhd.TrackID != 1 {
|
||||
continue
|
||||
}
|
||||
for _, trun := range traf.Truns {
|
||||
for _, s := range trun.Samples {
|
||||
dur += uint64(s.Dur)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
actual = append(actual, segInfo{size: sz, dur: dur})
|
||||
}
|
||||
for i := range actual {
|
||||
refStr := "(no sidx ref)"
|
||||
if i < len(sidxRefs) {
|
||||
r := sidxRefs[i]
|
||||
mark := ""
|
||||
if uint64(r.ReferencedSize) != actual[i].size {
|
||||
mark += fmt.Sprintf(" SIZE MISMATCH actual=%d", actual[i].size)
|
||||
}
|
||||
if uint64(r.SubSegmentDuration) != actual[i].dur {
|
||||
mark += fmt.Sprintf(" DUR MISMATCH actual=%d", actual[i].dur)
|
||||
}
|
||||
refStr = fmt.Sprintf("sidx.size=%d sidx.dur=%d type=%d sap=%d/%d%s",
|
||||
r.ReferencedSize, r.SubSegmentDuration, r.ReferenceType, r.StartsWithSAP, r.SAPType, mark)
|
||||
}
|
||||
fmt.Printf(" seg%02d actual.size=%d actual.dur=%d | %s\n", i, actual[i].size, actual[i].dur, refStr)
|
||||
}
|
||||
|
||||
fmt.Println("=== fragments ===")
|
||||
fragIdx := 0
|
||||
var allKeyGlobal []uint64 // global keyframe decode times (track timescale units)
|
||||
var prevTfdtEnd = map[uint32]uint64{}
|
||||
for si, seg := range parsed.Segments {
|
||||
for _, fr := range seg.Fragments {
|
||||
for _, traf := range fr.Moof.Trafs {
|
||||
tid := traf.Tfhd.TrackID
|
||||
tfdt := traf.Tfdt.BaseMediaDecodeTime()
|
||||
offset := uint64(0)
|
||||
var keys []uint64 // keyframe offset-from-tfdt
|
||||
var durs []uint64
|
||||
zeroDur := 0
|
||||
nSamples := 0
|
||||
for _, trun := range traf.Truns {
|
||||
for _, s := range trun.Samples {
|
||||
nSamples++
|
||||
if (s.Flags>>24)&0x03 == 0x02 { // sample_depends_on==2 => IDR/sync
|
||||
keys = append(keys, offset)
|
||||
if tid == 1 {
|
||||
allKeyGlobal = append(allKeyGlobal, tfdt+offset)
|
||||
}
|
||||
}
|
||||
if s.Dur == 0 {
|
||||
zeroDur++
|
||||
}
|
||||
durs = append(durs, uint64(s.Dur))
|
||||
offset += uint64(s.Dur)
|
||||
}
|
||||
}
|
||||
cont := ""
|
||||
if pe, ok := prevTfdtEnd[tid]; ok {
|
||||
if tfdt != pe {
|
||||
cont = fmt.Sprintf(" <-- tfdt GAP/JUMP prev_end=%d delta=%d", pe, int64(tfdt)-int64(pe))
|
||||
}
|
||||
}
|
||||
prevTfdtEnd[tid] = tfdt + offset
|
||||
if tid == 1 {
|
||||
// in-fragment keyframe gaps
|
||||
var gaps []int64
|
||||
for i := 1; i < len(keys); i++ {
|
||||
gaps = append(gaps, int64(keys[i])-int64(keys[i-1]))
|
||||
}
|
||||
fmt.Printf("seg%d frag%d trk%d tfdt=%d dur=%d nSamp=%d zeroDur=%d keys=%v inFragKeyGaps=%v%s\n",
|
||||
si, fragIdx, tid, tfdt, offset, nSamples, zeroDur, keys, gaps, cont)
|
||||
}
|
||||
}
|
||||
fragIdx++
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Println("=== global video keyframe decode times & gaps ===")
|
||||
for i, k := range allKeyGlobal {
|
||||
gap := int64(0)
|
||||
if i > 0 {
|
||||
gap = int64(k) - int64(allKeyGlobal[i-1])
|
||||
}
|
||||
flag := ""
|
||||
if i > 1 {
|
||||
prevGap := int64(allKeyGlobal[i-1]) - int64(allKeyGlobal[i-2])
|
||||
if gap > 0 && prevGap > 0 && gap*2 < prevGap {
|
||||
flag = fmt.Sprintf(" <== SEAM? gap=%d < prevGap/2=%d", gap, prevGap/2)
|
||||
}
|
||||
}
|
||||
fmt.Printf(" kf#%02d dt=%d gap=%d%s\n", i, k, gap, flag)
|
||||
}
|
||||
|
||||
// Full sample timeline: DTS, CTS (=DTS+cto), composition offset, NAL types,
|
||||
// to detect PTS non-monotonicity / gaps / param-set changes at the seam.
|
||||
fmt.Println("=== per-sample timeline (full) — checking PTS monotonicity & nal types ===")
|
||||
var trex *mp4ff.TrexBox
|
||||
if parsed.Init != nil && parsed.Init.Moov != nil && parsed.Init.Moov.Mvex != nil {
|
||||
for _, t := range parsed.Init.Moov.Mvex.Trexs {
|
||||
if t.TrackID == 1 {
|
||||
trex = t
|
||||
}
|
||||
}
|
||||
}
|
||||
var lastCTS int64 = -1
|
||||
var lastDTS int64 = -1
|
||||
sampIdx := 0
|
||||
fragIdx = 0
|
||||
for _, seg := range parsed.Segments {
|
||||
for _, fr := range seg.Fragments {
|
||||
fs, err := fr.GetFullSamples(trex)
|
||||
if err != nil {
|
||||
fmt.Printf(" frag%d GetFullSamples err: %v\n", fragIdx, err)
|
||||
fragIdx++
|
||||
continue
|
||||
}
|
||||
for _, s := range fs {
|
||||
dts := int64(s.DecodeTime)
|
||||
cts := dts + int64(s.CompositionTimeOffset)
|
||||
nals := nalTypes(s.Data)
|
||||
anomaly := ""
|
||||
if lastCTS >= 0 && cts < lastCTS {
|
||||
anomaly += fmt.Sprintf(" <== CTS BACKWARDS (prev=%d)", lastCTS)
|
||||
}
|
||||
if lastDTS >= 0 && dts < lastDTS {
|
||||
anomaly += fmt.Sprintf(" <== DTS BACKWARDS (prev=%d)", lastDTS)
|
||||
}
|
||||
isSync := s.Flags&0x02000000 == 0 && (s.Flags>>24)&0x03 == 0x02
|
||||
// Only print near the seam region and any anomalies, to keep output small.
|
||||
near := dts >= 7800 && dts <= 8700
|
||||
if near || anomaly != "" {
|
||||
fmt.Printf(" s%04d frag%d dts=%d cts=%d cto=%d dur=%d size=%d sync=%v nal=%v%s\n",
|
||||
sampIdx, fragIdx, dts, cts, s.CompositionTimeOffset, s.Dur, len(s.Data), isSync, nals, anomaly)
|
||||
}
|
||||
lastCTS = cts
|
||||
lastDTS = dts
|
||||
sampIdx++
|
||||
}
|
||||
fragIdx++
|
||||
}
|
||||
}
|
||||
|
||||
// Compare parameter sets: avcC (in moov) vs inline SPS/PPS at every IDR.
|
||||
// A looping source that restarts may re-emit SPS/PPS that differ from the
|
||||
// ones the player configured its decoder with from avcC — a classic cause
|
||||
// of a freeze that "heals" when you seek past the seam.
|
||||
fmt.Println("=== parameter set comparison (avcC vs inline IDR) ===")
|
||||
var avccSPS, avccPPS [][]byte
|
||||
if parsed.Init != nil && parsed.Init.Moov != nil {
|
||||
for _, trak := range parsed.Init.Moov.Traks {
|
||||
if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil {
|
||||
continue
|
||||
}
|
||||
stsd := trak.Mdia.Minf.Stbl.Stsd
|
||||
if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil {
|
||||
continue
|
||||
}
|
||||
avccSPS = stsd.AvcX.AvcC.SPSnalus
|
||||
avccPPS = stsd.AvcX.AvcC.PPSnalus
|
||||
}
|
||||
}
|
||||
for i, s := range avccSPS {
|
||||
fmt.Printf(" avcC SPS[%d] = %x\n", i, s)
|
||||
}
|
||||
for i, p := range avccPPS {
|
||||
fmt.Printf(" avcC PPS[%d] = %x\n", i, p)
|
||||
}
|
||||
fragIdx = 0
|
||||
sampIdx = 0
|
||||
var baseSPS, basePPS []byte
|
||||
if len(avccSPS) > 0 {
|
||||
baseSPS = avccSPS[0]
|
||||
}
|
||||
if len(avccPPS) > 0 {
|
||||
basePPS = avccPPS[0]
|
||||
}
|
||||
for _, seg := range parsed.Segments {
|
||||
for _, fr := range seg.Fragments {
|
||||
fs, err := fr.GetFullSamples(trex)
|
||||
if err != nil {
|
||||
fragIdx++
|
||||
continue
|
||||
}
|
||||
for _, s := range fs {
|
||||
spsList := nalsByType(s.Data, 7)
|
||||
ppsList := nalsByType(s.Data, 8)
|
||||
if len(spsList) > 0 || len(ppsList) > 0 {
|
||||
dts := int64(s.DecodeTime)
|
||||
note := ""
|
||||
if len(spsList) > 0 {
|
||||
if baseSPS == nil {
|
||||
baseSPS = spsList[0]
|
||||
} else if !bytesEqual(baseSPS, spsList[0]) {
|
||||
note += " <== SPS CHANGED vs base/avcC"
|
||||
}
|
||||
}
|
||||
if len(ppsList) > 0 {
|
||||
if basePPS == nil {
|
||||
basePPS = ppsList[0]
|
||||
} else if !bytesEqual(basePPS, ppsList[0]) {
|
||||
note += " <== PPS CHANGED vs base/avcC"
|
||||
}
|
||||
}
|
||||
var spsHex, ppsHex string
|
||||
if len(spsList) > 0 {
|
||||
spsHex = fmt.Sprintf("%x", spsList[0])
|
||||
}
|
||||
if len(ppsList) > 0 {
|
||||
ppsHex = fmt.Sprintf("%x", ppsList[0])
|
||||
}
|
||||
fmt.Printf(" IDR s%04d frag%d dts=%d SPS=%s PPS=%s%s\n",
|
||||
sampIdx, fragIdx, dts, spsHex, ppsHex, note)
|
||||
}
|
||||
sampIdx++
|
||||
}
|
||||
fragIdx++
|
||||
}
|
||||
}
|
||||
|
||||
sliceHeaders(parsed, trex)
|
||||
}
|
||||
|
||||
func sliceHeaders(parsed *mp4ff.File, trex *mp4ff.TrexBox) {
|
||||
// Build SPS/PPS maps from avcC.
|
||||
spsMap := map[uint32]*avc.SPS{}
|
||||
ppsMap := map[uint32]*avc.PPS{}
|
||||
if parsed.Init != nil && parsed.Init.Moov != nil {
|
||||
for _, trak := range parsed.Init.Moov.Traks {
|
||||
if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil {
|
||||
continue
|
||||
}
|
||||
stsd := trak.Mdia.Minf.Stbl.Stsd
|
||||
if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil {
|
||||
continue
|
||||
}
|
||||
for _, s := range stsd.AvcX.AvcC.SPSnalus {
|
||||
if sps, err := avc.ParseSPSNALUnit(s, true); err == nil {
|
||||
spsMap[uint32(sps.ParameterID)] = sps
|
||||
}
|
||||
}
|
||||
for _, p := range stsd.AvcX.AvcC.PPSnalus {
|
||||
if pps, err := avc.ParsePPSNALUnit(p, spsMap); err == nil {
|
||||
ppsMap[pps.PicParameterSetID] = pps
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Println("=== slice headers near seam (frame_num / poc / idr_pic_id) ===")
|
||||
fragIdx := 0
|
||||
sampIdx := 0
|
||||
for _, seg := range parsed.Segments {
|
||||
for _, fr := range seg.Fragments {
|
||||
fs, err := fr.GetFullSamples(trex)
|
||||
if err != nil {
|
||||
fragIdx++
|
||||
continue
|
||||
}
|
||||
for _, s := range fs {
|
||||
dts := int64(s.DecodeTime)
|
||||
if dts < 6800 || dts > 9400 {
|
||||
sampIdx++
|
||||
continue
|
||||
}
|
||||
for _, nal := range splitAVCC(s.Data) {
|
||||
t := nal[0] & 0x1f
|
||||
if t == 1 || t == 5 { // non-IDR or IDR slice
|
||||
sh, err := avc.ParseSliceHeader(nal, spsMap, ppsMap)
|
||||
if err != nil {
|
||||
fmt.Printf(" s%04d frag%d dts=%d nalType=%d sliceHeader ERR: %v\n", sampIdx, fragIdx, dts, t, err)
|
||||
break
|
||||
}
|
||||
fmt.Printf(" s%04d frag%d dts=%d nalType=%d sliceType=%v frameNum=%d idrPicId=%d pocLsb=%d\n",
|
||||
sampIdx, fragIdx, dts, t, sh.SliceType, sh.FrameNum, sh.IDRPicID, sh.PicOrderCntLsb)
|
||||
break
|
||||
}
|
||||
}
|
||||
sampIdx++
|
||||
}
|
||||
fragIdx++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// splitAVCC splits a length-prefixed (4-byte) AVCC buffer into NAL units.
|
||||
func splitAVCC(b []byte) [][]byte {
|
||||
var out [][]byte
|
||||
i := 0
|
||||
for i+4 <= len(b) {
|
||||
n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
|
||||
i += 4
|
||||
if n <= 0 || i+n > len(b) {
|
||||
break
|
||||
}
|
||||
out = append(out, b[i:i+n])
|
||||
i += n
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func bytesEqual(a, b []byte) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// nalTypes returns the list of H.264 NAL unit types present in an AVCC
|
||||
// (length-prefixed) sample buffer.
|
||||
func nalTypes(b []byte) []int {
|
||||
var out []int
|
||||
i := 0
|
||||
for i+4 <= len(b) {
|
||||
n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
|
||||
i += 4
|
||||
if n <= 0 || i+n > len(b) {
|
||||
break
|
||||
}
|
||||
out = append(out, int(b[i]&0x1f))
|
||||
i += n
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// nalsByType returns the raw NAL payloads (without length prefix) of the given
|
||||
// type from an AVCC (length-prefixed) sample buffer.
|
||||
func nalsByType(b []byte, want int) [][]byte {
|
||||
var out [][]byte
|
||||
i := 0
|
||||
for i+4 <= len(b) {
|
||||
n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
|
||||
i += 4
|
||||
if n <= 0 || i+n > len(b) {
|
||||
break
|
||||
}
|
||||
if int(b[i]&0x1f) == want {
|
||||
nal := make([]byte, n)
|
||||
copy(nal, b[i:i+n])
|
||||
out = append(out, nal)
|
||||
}
|
||||
i += n
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -30,6 +30,15 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
|
||||
return false, false, errors.New(err)
|
||||
}
|
||||
|
||||
// If the recording no longer exists on disk there is nothing to upload.
|
||||
// This can happen when the file was already removed (e.g. cleanup, or an
|
||||
// earlier successful upload). Skip it so the watcher drops the marker
|
||||
// instead of retrying indefinitely.
|
||||
if _, err := os.Stat("data/recordings/" + fileName); err != nil {
|
||||
log.Log.Info("UploadKerberosVault: skipping " + fileName + ", file doesn't exist anymore")
|
||||
return false, false, nil
|
||||
}
|
||||
|
||||
// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
|
||||
// 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4
|
||||
// - Timestamp
|
||||
@@ -41,17 +50,6 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
|
||||
// KerberosCloud, this means storage is disabled and proxy enabled.
|
||||
log.Log.Info("UploadKerberosVault: Uploading to Kerberos Vault (" + config.KStorage.URI + ")")
|
||||
log.Log.Info("UploadKerberosVault: Upload started for " + fileName)
|
||||
fullname := "data/recordings/" + fileName
|
||||
|
||||
file, err := os.OpenFile(fullname, os.O_RDWR, 0755)
|
||||
if file != nil {
|
||||
defer file.Close()
|
||||
}
|
||||
if err != nil {
|
||||
err := "UploadKerberosVault: Upload Failed, file doesn't exists anymore"
|
||||
log.Log.Info(err)
|
||||
return false, false, errors.New(err)
|
||||
}
|
||||
|
||||
publicKey := config.KStorage.CloudKey
|
||||
if config.HubKey != "" {
|
||||
@@ -60,62 +58,30 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
|
||||
|
||||
// We need to check if we are in a retry timeout.
|
||||
if kstorageRetryTimeout <= time.Now().Unix() {
|
||||
uploaded, responded, body, err := sendToVault(*config.KStorage, publicKey, config.Key, fileName, "UploadKerberosVault", "primary")
|
||||
if uploaded {
|
||||
kstorageRetryCount = 0
|
||||
log.Log.Info("UploadKerberosVault: Upload Finished, " + body)
|
||||
return true, true, nil
|
||||
}
|
||||
|
||||
req, err := http.NewRequest("POST", config.KStorage.URI+"/storage", file)
|
||||
if err != nil {
|
||||
errorMessage := "UploadKerberosVault: error reading request, " + config.KStorage.URI + "/storage: " + err.Error()
|
||||
log.Log.Error(errorMessage)
|
||||
return false, true, errors.New(errorMessage)
|
||||
}
|
||||
req.Header.Set("Content-Type", "video/mp4")
|
||||
req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey)
|
||||
req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorage.AccessKey)
|
||||
req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorage.SecretAccessKey)
|
||||
req.Header.Set("X-Kerberos-Storage-Provider", config.KStorage.Provider)
|
||||
req.Header.Set("X-Kerberos-Storage-FileName", fileName)
|
||||
req.Header.Set("X-Kerberos-Storage-Device", config.Key)
|
||||
req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera")
|
||||
req.Header.Set("X-Kerberos-Storage-Directory", config.KStorage.Directory)
|
||||
|
||||
var client *http.Client
|
||||
if os.Getenv("AGENT_TLS_INSECURE") == "true" {
|
||||
tr := &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
|
||||
}
|
||||
client = &http.Client{Transport: tr}
|
||||
} else {
|
||||
client = &http.Client{}
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
if resp != nil {
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err == nil {
|
||||
if resp.StatusCode == 200 {
|
||||
kstorageRetryCount = 0
|
||||
log.Log.Info("UploadKerberosVault: Upload Finished, " + resp.Status + ", " + string(body))
|
||||
return true, true, nil
|
||||
} else {
|
||||
// We increase the retry count, and set the timeout.
|
||||
// If we have reached the retry policy, we set the timeout.
|
||||
// This means we will not retry for the next 5 minutes.
|
||||
if kstorageRetryCount < config.KStorage.MaxRetries {
|
||||
kstorageRetryCount = (kstorageRetryCount + 1)
|
||||
}
|
||||
if kstorageRetryCount == config.KStorage.MaxRetries {
|
||||
kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix()
|
||||
}
|
||||
log.Log.Info("UploadKerberosVault: Upload Failed, " + resp.Status + ", " + string(body))
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
log.Log.Info("UploadKerberosVault: Upload Failed, " + err.Error())
|
||||
} else {
|
||||
log.Log.Info("UploadKerberosVault: Upload Failed, " + body)
|
||||
}
|
||||
|
||||
// We only advance the retry policy when the vault gave a definitive
|
||||
// response (mirroring the original behaviour where transient network
|
||||
// errors did not consume retries). When the retry count reaches the
|
||||
// configured maximum we back off for the configured timeout.
|
||||
if responded {
|
||||
if kstorageRetryCount < config.KStorage.MaxRetries {
|
||||
kstorageRetryCount = (kstorageRetryCount + 1)
|
||||
}
|
||||
if kstorageRetryCount == config.KStorage.MaxRetries {
|
||||
kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -134,61 +100,116 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
|
||||
|
||||
log.Log.Info("UploadKerberosVault (Secondary): Uploading to Secondary Kerberos Vault (" + config.KStorageSecondary.URI + ")")
|
||||
|
||||
file, err = os.OpenFile(fullname, os.O_RDWR, 0755)
|
||||
if file != nil {
|
||||
defer file.Close()
|
||||
}
|
||||
if err != nil {
|
||||
err := "UploadKerberosVault (Secondary): Upload Failed, file doesn't exists anymore"
|
||||
log.Log.Info(err)
|
||||
return false, false, errors.New(err)
|
||||
uploaded, _, body, err := sendToVault(*config.KStorageSecondary, publicKey, config.Key, fileName, "UploadKerberosVault (Secondary)", "secondary")
|
||||
if uploaded {
|
||||
log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + body)
|
||||
return true, true, nil
|
||||
}
|
||||
|
||||
req, err := http.NewRequest("POST", config.KStorageSecondary.URI+"/storage", file)
|
||||
if err != nil {
|
||||
errorMessage := "UploadKerberosVault (Secondary): error reading request, " + config.KStorageSecondary.URI + "/storage: " + err.Error()
|
||||
log.Log.Error(errorMessage)
|
||||
return false, true, errors.New(errorMessage)
|
||||
}
|
||||
req.Header.Set("Content-Type", "video/mp4")
|
||||
req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey)
|
||||
req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorageSecondary.AccessKey)
|
||||
req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorageSecondary.SecretAccessKey)
|
||||
req.Header.Set("X-Kerberos-Storage-Provider", config.KStorageSecondary.Provider)
|
||||
req.Header.Set("X-Kerberos-Storage-FileName", fileName)
|
||||
req.Header.Set("X-Kerberos-Storage-Device", config.Key)
|
||||
req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera")
|
||||
req.Header.Set("X-Kerberos-Storage-Directory", config.KStorageSecondary.Directory)
|
||||
|
||||
var client *http.Client
|
||||
if os.Getenv("AGENT_TLS_INSECURE") == "true" {
|
||||
tr := &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
|
||||
}
|
||||
client = &http.Client{Transport: tr}
|
||||
log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + err.Error())
|
||||
} else {
|
||||
client = &http.Client{}
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
if resp != nil {
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err == nil {
|
||||
if resp.StatusCode == 200 {
|
||||
log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + resp.Status + ", " + string(body))
|
||||
return true, true, nil
|
||||
} else {
|
||||
log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + resp.Status + ", " + string(body))
|
||||
}
|
||||
}
|
||||
}
|
||||
log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + body)
|
||||
}
|
||||
}
|
||||
|
||||
return false, true, nil
|
||||
}
|
||||
|
||||
// sendToVault uploads a single recording to one Kerberos Vault. When resumable
|
||||
// uploads are enabled (the default) it attempts the tus protocol first and, if
|
||||
// the vault does not expose a tus endpoint (older deployments), transparently
|
||||
// falls back to the legacy single-shot POST.
|
||||
//
|
||||
// It returns whether the upload succeeded, whether the vault gave a definitive
|
||||
// HTTP response (so the caller can advance its retry policy), a short message
|
||||
// for logging, and a transport error if any.
|
||||
func sendToVault(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, string, error) {
|
||||
if resumableUploadsEnabled() {
|
||||
uploaded, responded, supported, body, err := uploadVaultResumable(vault, publicKey, deviceKey, fileName, label, slot)
|
||||
if supported {
|
||||
return uploaded, responded, body, err
|
||||
}
|
||||
log.Log.Info(label + ": resumable (tus) endpoint not available, falling back to legacy upload")
|
||||
}
|
||||
return uploadVaultLegacy(vault, publicKey, deviceKey, fileName, label)
|
||||
}
|
||||
|
||||
// uploadVaultLegacy performs the original single-request upload: the whole file
|
||||
// is sent as the body of a POST to {URI}/storage. Kept for backwards
|
||||
// compatibility with vault deployments that do not support resumable uploads.
|
||||
func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, label string) (bool, bool, string, error) {
|
||||
fullname := "data/recordings/" + fileName
|
||||
|
||||
file, err := os.Open(fullname)
|
||||
if file != nil {
|
||||
defer file.Close()
|
||||
}
|
||||
if err != nil {
|
||||
msg := label + ": Upload Failed, file doesn't exists anymore"
|
||||
log.Log.Info(msg)
|
||||
return false, false, "", errors.New(msg)
|
||||
}
|
||||
|
||||
uri := vault.URI
|
||||
for len(uri) > 0 && uri[len(uri)-1] == '/' {
|
||||
uri = uri[:len(uri)-1]
|
||||
}
|
||||
|
||||
req, err := http.NewRequest("POST", uri+"/storage", file)
|
||||
if err != nil {
|
||||
errorMessage := label + ": error reading request, " + uri + "/storage: " + err.Error()
|
||||
log.Log.Error(errorMessage)
|
||||
return false, false, "", errors.New(errorMessage)
|
||||
}
|
||||
req.Header.Set("Content-Type", "video/mp4")
|
||||
setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
|
||||
|
||||
client := newVaultHTTPClient(0)
|
||||
resp, err := client.Do(req)
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
}
|
||||
if err != nil {
|
||||
return false, false, "", err
|
||||
}
|
||||
|
||||
body, rerr := io.ReadAll(resp.Body)
|
||||
if rerr != nil {
|
||||
return false, false, "", rerr
|
||||
}
|
||||
|
||||
if resp.StatusCode == 200 {
|
||||
return true, true, resp.Status + ", " + string(body), nil
|
||||
}
|
||||
return false, true, resp.Status + ", " + string(body), nil
|
||||
}
|
||||
|
||||
// setVaultHeaders sets the standard Kerberos Vault headers used by the legacy
|
||||
// single-POST upload.
|
||||
func setVaultHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
|
||||
h.Set("X-Kerberos-Storage-CloudKey", publicKey)
|
||||
h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
|
||||
h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
|
||||
h.Set("X-Kerberos-Storage-Provider", vault.Provider)
|
||||
h.Set("X-Kerberos-Storage-FileName", fileName)
|
||||
h.Set("X-Kerberos-Storage-Device", deviceKey)
|
||||
h.Set("X-Kerberos-Storage-Capture", "IPCamera")
|
||||
h.Set("X-Kerberos-Storage-Directory", vault.Directory)
|
||||
}
|
||||
|
||||
// newVaultHTTPClient builds an HTTP client honouring the AGENT_TLS_INSECURE
|
||||
// escape hatch. A timeout of 0 disables the client-level timeout, which is
|
||||
// required for streaming large upload bodies.
|
||||
func newVaultHTTPClient(timeout time.Duration) *http.Client {
|
||||
client := &http.Client{}
|
||||
if os.Getenv("AGENT_TLS_INSECURE") == "true" {
|
||||
client.Transport = &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
|
||||
}
|
||||
}
|
||||
if timeout > 0 {
|
||||
client.Timeout = timeout
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
452
machinery/src/cloud/tus_client.go
Normal file
452
machinery/src/cloud/tus_client.go
Normal file
@@ -0,0 +1,452 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/kerberos-io/agent/machinery/src/log"
|
||||
"github.com/kerberos-io/agent/machinery/src/models"
|
||||
)
|
||||
|
||||
// tusResumableVersion is the tus protocol version implemented by this client.
|
||||
const tusResumableVersion = "1.0.0"
|
||||
|
||||
// tusUploadPath is appended to the configured Kerberos Vault URI to reach the
|
||||
// resumable upload endpoint. It mirrors how the legacy uploader appends
|
||||
// "/storage".
|
||||
const tusUploadPath = "/storage/tus/"
|
||||
|
||||
// tusResumeState is persisted in a sidecar file next to the agent data so an
|
||||
// interrupted upload can be resumed across retries and even agent restarts.
|
||||
type tusResumeState struct {
|
||||
UploadURL string `json:"upload_url"`
|
||||
VaultURI string `json:"vault_uri"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
|
||||
// resumableUploadsEnabled reports whether the resumable (tus) upload path should
|
||||
// be attempted. It is enabled by default and can be disabled (falling back to
|
||||
// the legacy single POST) by setting AGENT_DISABLE_RESUMABLE_UPLOAD=true.
|
||||
func resumableUploadsEnabled() bool {
|
||||
return os.Getenv("AGENT_DISABLE_RESUMABLE_UPLOAD") != "true"
|
||||
}
|
||||
|
||||
// tusDefaultChunkSize is the number of bytes uploaded per PATCH request when no
|
||||
// explicit size is configured. Splitting the upload into chunks keeps each HTTP
|
||||
// request small enough for intermediary proxies/load balancers and checkpoints
|
||||
// progress frequently, so an interruption resumes with minimal re-upload.
|
||||
const tusDefaultChunkSize int64 = 1 << 20 // 1 MiB
|
||||
|
||||
// tusChunkSize returns the number of bytes to send per PATCH request. It
|
||||
// defaults to tusDefaultChunkSize (1 MiB) and can be overridden with the
|
||||
// AGENT_TUS_CHUNK_SIZE_BYTES environment variable. A value of 0 (or negative)
|
||||
// disables chunking and sends the remaining bytes in a single PATCH.
|
||||
func tusChunkSize() int64 {
|
||||
v := os.Getenv("AGENT_TUS_CHUNK_SIZE_BYTES")
|
||||
if v == "" {
|
||||
return tusDefaultChunkSize
|
||||
}
|
||||
n, err := strconv.ParseInt(v, 10, 64)
|
||||
if err != nil {
|
||||
return tusDefaultChunkSize
|
||||
}
|
||||
if n <= 0 {
|
||||
return 0 // chunking disabled: send everything in one PATCH
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// uploadVaultResumable uploads a recording to a Kerberos Vault using the tus
|
||||
// resumable upload protocol.
|
||||
//
|
||||
// Return values:
|
||||
// - uploaded: the recording was fully received and persisted by the vault.
|
||||
// - responded: the vault returned a definitive HTTP response (used by the
|
||||
// caller to advance its retry/secondary-failover policy).
|
||||
// - supported: the vault exposes a tus endpoint. When false, the caller should
|
||||
// fall back to the legacy single-POST upload (older vault deployments).
|
||||
// - body: a short message for logging.
|
||||
func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (uploaded bool, responded bool, supported bool, body string, err error) {
|
||||
fullname := "data/recordings/" + fileName
|
||||
|
||||
file, ferr := os.Open(fullname)
|
||||
if file != nil {
|
||||
defer file.Close()
|
||||
}
|
||||
if ferr != nil {
|
||||
msg := label + ": resumable upload failed, file doesn't exist anymore"
|
||||
log.Log.Info(msg)
|
||||
// The file is gone, so the legacy path cannot help either. Report it as
|
||||
// "supported" to avoid a pointless fallback attempt.
|
||||
return false, false, true, "", errors.New(msg)
|
||||
}
|
||||
|
||||
info, serr := file.Stat()
|
||||
if serr != nil {
|
||||
return false, false, true, "", serr
|
||||
}
|
||||
size := info.Size()
|
||||
|
||||
baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
|
||||
client := newVaultHTTPClient(0)
|
||||
|
||||
metadata := encodeTusMetadata(map[string]string{
|
||||
"filename": fileName,
|
||||
"device": deviceKey,
|
||||
"directory": vault.Directory,
|
||||
"provider": vault.Provider,
|
||||
"capture": "IPCamera",
|
||||
"cloudkey": publicKey,
|
||||
})
|
||||
|
||||
sidecar := tusSidecarPath(fileName, slot)
|
||||
uploadURL := loadTusResumeState(sidecar, baseURL)
|
||||
|
||||
const maxAttempts = 4
|
||||
restartedAfterComplete := false
|
||||
|
||||
for attempt := 0; attempt < maxAttempts; attempt++ {
|
||||
// (1) Ensure we have an active upload URL, creating one if needed.
|
||||
if uploadURL == "" {
|
||||
created, status, cerr := tusCreate(client, baseURL, size, metadata, vault, publicKey, deviceKey, fileName)
|
||||
if cerr != nil {
|
||||
if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
|
||||
// The vault does not implement tus; let the caller fall back.
|
||||
return false, false, false, "", cerr
|
||||
}
|
||||
log.Log.Info(label + ": resumable create failed, " + cerr.Error())
|
||||
tusBackoff(attempt)
|
||||
continue
|
||||
}
|
||||
uploadURL = created
|
||||
saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
|
||||
}
|
||||
|
||||
// (2) Query the current server-side offset.
|
||||
offset, status, herr := tusHead(client, uploadURL, vault, publicKey, deviceKey)
|
||||
if herr != nil {
|
||||
if status == http.StatusNotFound || status == http.StatusGone {
|
||||
// The upload expired/was removed server-side; start over.
|
||||
removeTusResumeState(sidecar)
|
||||
uploadURL = ""
|
||||
continue
|
||||
}
|
||||
log.Log.Info(label + ": resumable head failed, " + herr.Error())
|
||||
tusBackoff(attempt)
|
||||
continue
|
||||
}
|
||||
|
||||
// (3) All bytes are present but the upload was not finalized (e.g. the
|
||||
// completion hook failed). A completed tus upload cannot be re-finalized
|
||||
// with another PATCH, so delete it and re-upload to force a clean finalize.
|
||||
if offset >= size {
|
||||
if restartedAfterComplete {
|
||||
return false, true, true, "resumable finalize did not complete", errors.New(label + ": resumable finalize did not complete")
|
||||
}
|
||||
tusTerminate(client, uploadURL, vault, publicKey, deviceKey)
|
||||
removeTusResumeState(sidecar)
|
||||
uploadURL = ""
|
||||
restartedAfterComplete = true
|
||||
continue
|
||||
}
|
||||
|
||||
// (4) Stream the remaining bytes to the vault via PATCH, reading directly
|
||||
// from disk so the recording is never fully buffered in memory. When a chunk
|
||||
// size is configured the data is sent across several PATCH requests,
|
||||
// checkpointing the offset after each one so an interruption resumes from the
|
||||
// last completed chunk instead of re-uploading everything.
|
||||
chunkSize := tusChunkSize()
|
||||
progressed := false
|
||||
patchFailed := false
|
||||
var lastBody string
|
||||
for offset < size {
|
||||
// Re-seek every chunk so the on-disk position always matches the
|
||||
// server-acknowledged offset, even if a PATCH was partially accepted.
|
||||
if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil {
|
||||
return false, false, true, "", sErr
|
||||
}
|
||||
patchLen := size - offset
|
||||
if chunkSize > 0 && chunkSize < patchLen {
|
||||
patchLen = chunkSize
|
||||
}
|
||||
newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, vault, publicKey, deviceKey)
|
||||
if perr != nil {
|
||||
if status >= 400 {
|
||||
// Definitive rejection (e.g. provider push failed during finalize).
|
||||
// Re-evaluate via HEAD on the next iteration to decide retry/restart.
|
||||
log.Log.Info(label + ": resumable patch rejected, " + perr.Error())
|
||||
} else {
|
||||
log.Log.Info(label + ": resumable patch failed, " + perr.Error())
|
||||
}
|
||||
tusBackoff(attempt)
|
||||
patchFailed = true
|
||||
break
|
||||
}
|
||||
if newOffset > offset {
|
||||
progressed = true
|
||||
}
|
||||
offset = newOffset
|
||||
lastBody = respBody
|
||||
if offset < size {
|
||||
// Partial progress: persist so a later retry resumes from here.
|
||||
saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
|
||||
}
|
||||
}
|
||||
if patchFailed {
|
||||
if progressed {
|
||||
// Forward progress refreshes the retry budget: maxAttempts bounds the
|
||||
// number of consecutive failures, not the number of chunks needed for
|
||||
// a large recording.
|
||||
attempt = -1
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// All declared bytes have been sent and acknowledged: the upload is done.
|
||||
removeTusResumeState(sidecar)
|
||||
return true, true, true, lastBody, nil
|
||||
}
|
||||
|
||||
return false, true, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
|
||||
}
|
||||
|
||||
// tusCreate performs the tus "creation" request (POST). On success it returns
|
||||
// the resolved upload URL the agent should use for subsequent HEAD/PATCH calls.
|
||||
func tusCreate(client *http.Client, baseURL string, size int64, metadata string, vault models.KStorage, publicKey, deviceKey, fileName string) (string, int, error) {
|
||||
req, err := http.NewRequest("POST", baseURL, nil)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
req.Header.Set("Tus-Resumable", tusResumableVersion)
|
||||
req.Header.Set("Upload-Length", strconv.FormatInt(size, 10))
|
||||
if metadata != "" {
|
||||
req.Header.Set("Upload-Metadata", metadata)
|
||||
}
|
||||
setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, fileName)
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
}
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
|
||||
if resp.StatusCode != http.StatusCreated {
|
||||
return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status)
|
||||
}
|
||||
location := resp.Header.Get("Location")
|
||||
if location == "" {
|
||||
return "", resp.StatusCode, errors.New("missing Location header in create response")
|
||||
}
|
||||
return resolveTusLocation(baseURL, location), resp.StatusCode, nil
|
||||
}
|
||||
|
||||
// tusHead performs the tus "offset" request (HEAD) and returns the current
|
||||
// server-side upload offset.
|
||||
func tusHead(client *http.Client, uploadURL string, vault models.KStorage, publicKey, deviceKey string) (int64, int, error) {
|
||||
req, err := http.NewRequest("HEAD", uploadURL, nil)
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
req.Header.Set("Tus-Resumable", tusResumableVersion)
|
||||
setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
}
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
|
||||
return 0, resp.StatusCode, fmt.Errorf("unexpected status on HEAD: %s", resp.Status)
|
||||
}
|
||||
offsetStr := resp.Header.Get("Upload-Offset")
|
||||
offset, perr := strconv.ParseInt(offsetStr, 10, 64)
|
||||
if perr != nil {
|
||||
return 0, resp.StatusCode, fmt.Errorf("invalid Upload-Offset header: %q", offsetStr)
|
||||
}
|
||||
return offset, resp.StatusCode, nil
|
||||
}
|
||||
|
||||
// tusPatch streams up to length bytes of the file (starting at offset) to the
|
||||
// upload URL using a single PATCH request. The body is read straight from the
|
||||
// *os.File, so the recording is never fully buffered in memory.
|
||||
func tusPatch(client *http.Client, uploadURL string, offset, length int64, file io.Reader, vault models.KStorage, publicKey, deviceKey string) (int64, int, string, error) {
|
||||
req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, length))
|
||||
if err != nil {
|
||||
return offset, 0, "", err
|
||||
}
|
||||
req.ContentLength = length
|
||||
req.Header.Set("Tus-Resumable", tusResumableVersion)
|
||||
req.Header.Set("Content-Type", "application/offset+octet-stream")
|
||||
req.Header.Set("Upload-Offset", strconv.FormatInt(offset, 10))
|
||||
setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
}
|
||||
if err != nil {
|
||||
return offset, 0, "", err
|
||||
}
|
||||
bodyBytes, _ := io.ReadAll(resp.Body)
|
||||
respBody := string(bodyBytes)
|
||||
|
||||
if resp.StatusCode != http.StatusNoContent {
|
||||
return offset, resp.StatusCode, respBody, fmt.Errorf("unexpected status on PATCH: %s, %s", resp.Status, respBody)
|
||||
}
|
||||
newOffsetStr := resp.Header.Get("Upload-Offset")
|
||||
newOffset, perr := strconv.ParseInt(newOffsetStr, 10, 64)
|
||||
if perr != nil {
|
||||
// A 204 without a parseable offset means this PATCH was fully accepted.
|
||||
return offset + length, resp.StatusCode, respBody, nil
|
||||
}
|
||||
return newOffset, resp.StatusCode, respBody, nil
|
||||
}
|
||||
|
||||
// tusTerminate best-effort deletes an upload server-side (DELETE).
|
||||
func tusTerminate(client *http.Client, uploadURL string, vault models.KStorage, publicKey, deviceKey string) {
|
||||
req, err := http.NewRequest("DELETE", uploadURL, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
req.Header.Set("Tus-Resumable", tusResumableVersion)
|
||||
setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "")
|
||||
|
||||
resp, derr := client.Do(req)
|
||||
if resp != nil {
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
resp.Body.Close()
|
||||
}
|
||||
_ = derr
|
||||
}
|
||||
|
||||
// setVaultTusHeaders sets the Kerberos Vault authentication and routing headers
|
||||
// on every tus request. Credentials are sent on each request (and never stored
|
||||
// server-side in the upload metadata). When fileName is empty it is omitted, as
|
||||
// it is only useful on the creation request (routing also travels in the tus
|
||||
// Upload-Metadata).
|
||||
func setVaultTusHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
|
||||
h.Set("X-Kerberos-Storage-CloudKey", publicKey)
|
||||
h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
|
||||
h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
|
||||
h.Set("X-Kerberos-Storage-Provider", vault.Provider)
|
||||
h.Set("X-Kerberos-Storage-Device", deviceKey)
|
||||
h.Set("X-Kerberos-Storage-Directory", vault.Directory)
|
||||
h.Set("X-Kerberos-Storage-Capture", "IPCamera")
|
||||
if fileName != "" {
|
||||
h.Set("X-Kerberos-Storage-FileName", fileName)
|
||||
}
|
||||
}
|
||||
|
||||
// encodeTusMetadata serializes a map into the tus Upload-Metadata header format:
|
||||
// a comma separated list of "key base64(value)" pairs. Keys are sorted for a
|
||||
// deterministic header value. Empty values are skipped.
|
||||
func encodeTusMetadata(pairs map[string]string) string {
|
||||
parts := make([]string, 0, len(pairs))
|
||||
for k, v := range pairs {
|
||||
if v == "" {
|
||||
continue
|
||||
}
|
||||
parts = append(parts, k+" "+base64.StdEncoding.EncodeToString([]byte(v)))
|
||||
}
|
||||
sort.Strings(parts)
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
|
||||
// resolveTusLocation turns the Location header returned by the create request
|
||||
// into an absolute URL. To keep talking to the agent's configured vault host
|
||||
// (and avoid issues when the vault sits behind a proxy that rewrites the host),
|
||||
// it keeps the configured base URL and only appends the server-assigned upload
|
||||
// id taken from the Location.
|
||||
func resolveTusLocation(baseURL, location string) string {
|
||||
if ref, err := url.Parse(location); err == nil {
|
||||
trimmed := strings.Trim(ref.Path, "/")
|
||||
if trimmed != "" {
|
||||
segments := strings.Split(trimmed, "/")
|
||||
id := segments[len(segments)-1]
|
||||
if id != "" {
|
||||
return strings.TrimRight(baseURL, "/") + "/" + id
|
||||
}
|
||||
}
|
||||
}
|
||||
// Fallback: resolve the reference against the base URL as-is.
|
||||
if base, err := url.Parse(baseURL); err == nil {
|
||||
if ref, err := url.Parse(location); err == nil {
|
||||
return base.ResolveReference(ref).String()
|
||||
}
|
||||
}
|
||||
return location
|
||||
}
|
||||
|
||||
// tusSidecarDir is the directory where resume state files are kept. It is
|
||||
// intentionally separate from data/cloud (which is scanned for recordings to
|
||||
// upload) so the sidecar files are never mistaken for recordings.
|
||||
func tusSidecarDir() string {
|
||||
return "data/tus"
|
||||
}
|
||||
|
||||
func tusSidecarPath(fileName, slot string) string {
|
||||
safe := strings.ReplaceAll(fileName, "/", "_")
|
||||
safe = strings.ReplaceAll(safe, string(os.PathSeparator), "_")
|
||||
return filepath.Join(tusSidecarDir(), safe+"."+slot+".json")
|
||||
}
|
||||
|
||||
// loadTusResumeState returns a previously stored upload URL for the given
|
||||
// sidecar, but only if it was created against the same vault base URL. Any
|
||||
// mismatch or read/parse error yields an empty string (start fresh).
|
||||
func loadTusResumeState(path, baseURL string) string {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
var state tusResumeState
|
||||
if err := json.Unmarshal(b, &state); err != nil {
|
||||
return ""
|
||||
}
|
||||
if state.UploadURL == "" || state.VaultURI != baseURL {
|
||||
return ""
|
||||
}
|
||||
return state.UploadURL
|
||||
}
|
||||
|
||||
func saveTusResumeState(path string, state tusResumeState) {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return
|
||||
}
|
||||
b, err := json.Marshal(state)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_ = os.WriteFile(path, b, 0o644)
|
||||
}
|
||||
|
||||
func removeTusResumeState(path string) {
|
||||
_ = os.Remove(path)
|
||||
}
|
||||
|
||||
// tusBackoff sleeps for an exponentially increasing duration (capped) between
|
||||
// resume attempts to avoid hammering a temporarily unavailable vault.
|
||||
func tusBackoff(attempt int) {
|
||||
delay := time.Duration(500*(1<<uint(attempt))) * time.Millisecond
|
||||
if delay > 3*time.Second {
|
||||
delay = 3 * time.Second
|
||||
}
|
||||
time.Sleep(delay)
|
||||
}
|
||||
448
machinery/src/cloud/tus_client_test.go
Normal file
448
machinery/src/cloud/tus_client_test.go
Normal file
@@ -0,0 +1,448 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/kerberos-io/agent/machinery/src/models"
|
||||
)
|
||||
|
||||
// fakeUpload tracks the state of a single resumable upload on the fake server.
|
||||
type fakeUpload struct {
|
||||
size int64
|
||||
offset int64
|
||||
}
|
||||
|
||||
// fakeTus is a tiny in-memory implementation of the tus 1.0.0 server protocol,
|
||||
// sufficient to exercise the agent's resumable client.
|
||||
type fakeTus struct {
|
||||
mu sync.Mutex
|
||||
uploads map[string]*fakeUpload
|
||||
counter int
|
||||
creates int
|
||||
lastPatchBytes int64
|
||||
patchSizes []int64
|
||||
|
||||
// unsupported makes the creation endpoint return 404, simulating an older
|
||||
// vault without a tus endpoint.
|
||||
unsupported bool
|
||||
// failFinalize causes the next N completing PATCH requests to return 502
|
||||
// after storing the bytes, simulating a failed completion hook.
|
||||
failFinalize int
|
||||
}
|
||||
|
||||
func newFakeTus() *fakeTus {
|
||||
return &fakeTus{uploads: map[string]*fakeUpload{}}
|
||||
}
|
||||
|
||||
func (s *fakeTus) seed(size, offset int64) string {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.counter++
|
||||
id := fmt.Sprintf("seed-%d", s.counter)
|
||||
s.uploads[id] = &fakeUpload{size: size, offset: offset}
|
||||
return id
|
||||
}
|
||||
|
||||
func (s *fakeTus) totalBytes() int64 {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var total int64
|
||||
for _, u := range s.uploads {
|
||||
total += u.offset
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
func (s *fakeTus) lastPatch() int64 {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.lastPatchBytes
|
||||
}
|
||||
|
||||
// patchCounts returns the number of PATCH requests received and the size of each.
|
||||
func (s *fakeTus) patchCounts() (int, []int64) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
sizes := make([]int64, len(s.patchSizes))
|
||||
copy(sizes, s.patchSizes)
|
||||
return len(s.patchSizes), sizes
|
||||
}
|
||||
|
||||
func (s *fakeTus) createCount() int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.creates
|
||||
}
|
||||
|
||||
func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
id := strings.TrimPrefix(r.URL.Path, tusUploadPath)
|
||||
w.Header().Set("Tus-Resumable", tusResumableVersion)
|
||||
|
||||
switch r.Method {
|
||||
case http.MethodPost:
|
||||
if s.unsupported {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
length, _ := strconv.ParseInt(r.Header.Get("Upload-Length"), 10, 64)
|
||||
s.mu.Lock()
|
||||
s.counter++
|
||||
s.creates++
|
||||
newID := fmt.Sprintf("up-%d", s.counter)
|
||||
s.uploads[newID] = &fakeUpload{size: length}
|
||||
s.mu.Unlock()
|
||||
w.Header().Set("Location", tusUploadPath+newID)
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
|
||||
case http.MethodHead:
|
||||
s.mu.Lock()
|
||||
u, ok := s.uploads[id]
|
||||
s.mu.Unlock()
|
||||
if !ok {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Upload-Offset", strconv.FormatInt(u.offset, 10))
|
||||
w.Header().Set("Upload-Length", strconv.FormatInt(u.size, 10))
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
case http.MethodPatch:
|
||||
s.mu.Lock()
|
||||
u, ok := s.uploads[id]
|
||||
s.mu.Unlock()
|
||||
if !ok {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
n, _ := io.Copy(io.Discard, r.Body)
|
||||
s.mu.Lock()
|
||||
u.offset += n
|
||||
s.lastPatchBytes = n
|
||||
s.patchSizes = append(s.patchSizes, n)
|
||||
complete := u.offset >= u.size
|
||||
failNow := complete && s.failFinalize > 0
|
||||
if failNow {
|
||||
s.failFinalize--
|
||||
}
|
||||
offset := u.offset
|
||||
s.mu.Unlock()
|
||||
|
||||
w.Header().Set("Upload-Offset", strconv.FormatInt(offset, 10))
|
||||
if failNow {
|
||||
// Bytes are stored but the (simulated) completion hook failed.
|
||||
w.WriteHeader(http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
|
||||
case http.MethodDelete:
|
||||
s.mu.Lock()
|
||||
delete(s.uploads, id)
|
||||
s.mu.Unlock()
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
|
||||
default:
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
}
|
||||
}
|
||||
|
||||
// withRecording switches into a fresh temp working directory containing a
|
||||
// recording at data/recordings/<fileName>. The working directory is restored on
|
||||
// cleanup. Tests using this helper must not run in parallel.
|
||||
func withRecording(t *testing.T, fileName string, payload []byte) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
old, err := os.Getwd()
|
||||
if err != nil {
|
||||
t.Fatalf("getwd: %v", err)
|
||||
}
|
||||
if err := os.Chdir(dir); err != nil {
|
||||
t.Fatalf("chdir: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.Chdir(old) })
|
||||
|
||||
if err := os.MkdirAll("data/recordings", 0o755); err != nil {
|
||||
t.Fatalf("mkdir recordings: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join("data/recordings", fileName), payload, 0o644); err != nil {
|
||||
t.Fatalf("write recording: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func testVault(uri string) models.KStorage {
|
||||
return models.KStorage{
|
||||
URI: uri,
|
||||
AccessKey: "ak",
|
||||
SecretAccessKey: "sk",
|
||||
Provider: "gcp",
|
||||
Directory: "dir",
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadVaultResumable_HappyPath(t *testing.T) {
|
||||
srv := newFakeTus()
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
|
||||
payload := bytes.Repeat([]byte("x"), 4096)
|
||||
withRecording(t, fileName, payload)
|
||||
|
||||
uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !uploaded || !responded || !supported {
|
||||
t.Fatalf("uploaded/responded/supported = %v/%v/%v, want all true", uploaded, responded, supported)
|
||||
}
|
||||
if got := srv.totalBytes(); got != int64(len(payload)) {
|
||||
t.Fatalf("server received %d bytes, want %d", got, len(payload))
|
||||
}
|
||||
if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
|
||||
t.Fatalf("expected sidecar to be removed after success, stat err = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadVaultResumable_Chunked(t *testing.T) {
|
||||
srv := newFakeTus()
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
|
||||
// 10 KiB payload uploaded in 4 KiB chunks => 3 PATCH requests (4096+4096+2048).
|
||||
payload := bytes.Repeat([]byte("c"), 10240)
|
||||
withRecording(t, fileName, payload)
|
||||
t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "4096")
|
||||
|
||||
uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !uploaded || !supported {
|
||||
t.Fatalf("expected chunked upload success, got uploaded=%v supported=%v", uploaded, supported)
|
||||
}
|
||||
if got := srv.totalBytes(); got != int64(len(payload)) {
|
||||
t.Fatalf("server received %d bytes, want %d", got, len(payload))
|
||||
}
|
||||
count, sizes := srv.patchCounts()
|
||||
if count != 3 {
|
||||
t.Fatalf("expected 3 chunked PATCH requests, got %d (sizes=%v)", count, sizes)
|
||||
}
|
||||
want := []int64{4096, 4096, 2048}
|
||||
for i, w := range want {
|
||||
if sizes[i] != w {
|
||||
t.Fatalf("chunk %d size = %d, want %d (sizes=%v)", i, sizes[i], w, sizes)
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
|
||||
t.Fatalf("expected sidecar removed after success, stat err = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadVaultResumable_ChunkingDisabled(t *testing.T) {
|
||||
srv := newFakeTus()
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
|
||||
payload := bytes.Repeat([]byte("d"), 10240)
|
||||
withRecording(t, fileName, payload)
|
||||
// 0 disables chunking: the whole file should go out in a single PATCH.
|
||||
t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "0")
|
||||
|
||||
uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !uploaded || !supported {
|
||||
t.Fatalf("expected success, got uploaded=%v supported=%v", uploaded, supported)
|
||||
}
|
||||
count, sizes := srv.patchCounts()
|
||||
if count != 1 {
|
||||
t.Fatalf("expected a single PATCH when chunking is disabled, got %d (sizes=%v)", count, sizes)
|
||||
}
|
||||
if sizes[0] != int64(len(payload)) {
|
||||
t.Fatalf("single PATCH size = %d, want %d", sizes[0], len(payload))
|
||||
}
|
||||
}
|
||||
|
||||
func TestTusChunkSize(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
env string
|
||||
set bool
|
||||
want int64
|
||||
}{
|
||||
{name: "default when unset", set: false, want: tusDefaultChunkSize},
|
||||
{name: "default on invalid", env: "notanumber", set: true, want: tusDefaultChunkSize},
|
||||
{name: "explicit value", env: "65536", set: true, want: 65536},
|
||||
{name: "zero disables", env: "0", set: true, want: 0},
|
||||
{name: "negative disables", env: "-5", set: true, want: 0},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if tc.set {
|
||||
t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", tc.env)
|
||||
} else {
|
||||
t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "")
|
||||
}
|
||||
if got := tusChunkSize(); got != tc.want {
|
||||
t.Fatalf("tusChunkSize() = %d, want %d", got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadVaultResumable_Unsupported(t *testing.T) {
|
||||
srv := newFakeTus()
|
||||
srv.unsupported = true
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
fileName := "f.mp4"
|
||||
withRecording(t, fileName, []byte("hello"))
|
||||
|
||||
uploaded, _, supported, _, _ := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
|
||||
if uploaded {
|
||||
t.Fatal("expected uploaded=false against a vault without a tus endpoint")
|
||||
}
|
||||
if supported {
|
||||
t.Fatal("expected supported=false so the caller falls back to the legacy upload")
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
|
||||
srv := newFakeTus()
|
||||
srv.failFinalize = 1
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
|
||||
payload := bytes.Repeat([]byte("y"), 2048)
|
||||
withRecording(t, fileName, payload)
|
||||
|
||||
uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !uploaded || !supported {
|
||||
t.Fatalf("expected success after a failed finalize + restart, got uploaded=%v supported=%v", uploaded, supported)
|
||||
}
|
||||
if got := srv.createCount(); got < 2 {
|
||||
t.Fatalf("expected at least 2 create requests (restart after failed finalize), got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadVaultResumable_ResumeFromSidecar(t *testing.T) {
|
||||
srv := newFakeTus()
|
||||
ts := httptest.NewServer(srv)
|
||||
defer ts.Close()
|
||||
|
||||
fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
|
||||
total := 8192
|
||||
half := 4096
|
||||
payload := bytes.Repeat([]byte("z"), total)
|
||||
withRecording(t, fileName, payload)
|
||||
|
||||
// Simulate a previous run that uploaded half the file before being interrupted.
|
||||
id := srv.seed(int64(total), int64(half))
|
||||
baseURL := strings.TrimRight(ts.URL, "/") + tusUploadPath
|
||||
saveTusResumeState(tusSidecarPath(fileName, "primary"), tusResumeState{
|
||||
UploadURL: strings.TrimRight(baseURL, "/") + "/" + id,
|
||||
VaultURI: baseURL,
|
||||
Size: int64(total),
|
||||
})
|
||||
|
||||
uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !uploaded || !supported {
|
||||
t.Fatalf("expected resume success, got uploaded=%v supported=%v", uploaded, supported)
|
||||
}
|
||||
if got := srv.lastPatch(); got != int64(total-half) {
|
||||
t.Fatalf("resume should only send the remaining %d bytes, sent %d", total-half, got)
|
||||
}
|
||||
if srv.createCount() != 0 {
|
||||
t.Fatalf("resume should not create a new upload, got %d creates", srv.createCount())
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeTusMetadata(t *testing.T) {
|
||||
got := encodeTusMetadata(map[string]string{
|
||||
"b": "2",
|
||||
"a": "1",
|
||||
"empty": "",
|
||||
})
|
||||
// keys sorted, empty values skipped, values base64-encoded.
|
||||
want := "a MQ==,b Mg=="
|
||||
if got != want {
|
||||
t.Fatalf("encodeTusMetadata = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveTusLocation(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
base string
|
||||
location string
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "absolute path location",
|
||||
base: "http://host/storage/tus/",
|
||||
location: "/storage/tus/abc",
|
||||
want: "http://host/storage/tus/abc",
|
||||
},
|
||||
{
|
||||
name: "absolute url keeps configured host",
|
||||
base: "http://host/storage/tus/",
|
||||
location: "http://internal:8080/storage/tus/xyz",
|
||||
want: "http://host/storage/tus/xyz",
|
||||
},
|
||||
{
|
||||
name: "relative id",
|
||||
base: "http://host/api/storage/tus/",
|
||||
location: "abc",
|
||||
want: "http://host/api/storage/tus/abc",
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := resolveTusLocation(tc.base, tc.location); got != tc.want {
|
||||
t.Fatalf("resolveTusLocation(%q, %q) = %q, want %q", tc.base, tc.location, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTusResumeStateRoundTrip(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
old, _ := os.Getwd()
|
||||
if err := os.Chdir(dir); err != nil {
|
||||
t.Fatalf("chdir: %v", err)
|
||||
}
|
||||
defer os.Chdir(old)
|
||||
|
||||
path := tusSidecarPath("file.mp4", "primary")
|
||||
state := tusResumeState{UploadURL: "http://host/storage/tus/abc", VaultURI: "http://host/storage/tus/", Size: 123}
|
||||
saveTusResumeState(path, state)
|
||||
|
||||
if got := loadTusResumeState(path, state.VaultURI); got != state.UploadURL {
|
||||
t.Fatalf("loadTusResumeState = %q, want %q", got, state.UploadURL)
|
||||
}
|
||||
// A mismatched vault URI must not be reused.
|
||||
if got := loadTusResumeState(path, "http://other/storage/tus/"); got != "" {
|
||||
t.Fatalf("loadTusResumeState with mismatched vault = %q, want empty", got)
|
||||
}
|
||||
}
|
||||
@@ -96,7 +96,7 @@ func NewAACTranscoder() (*AACTranscoder, error) {
|
||||
buffered := t.outBuf.Len()
|
||||
t.outMu.Unlock()
|
||||
if buffered <= 8192 || buffered%16000 == 0 {
|
||||
log.Log.Info("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered))
|
||||
log.Log.Debug("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered))
|
||||
}
|
||||
}
|
||||
if readErr != nil {
|
||||
@@ -129,14 +129,14 @@ func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) {
|
||||
return nil, err
|
||||
}
|
||||
if len(adtsData) <= 512 || len(adtsData)%1024 == 0 {
|
||||
log.Log.Info("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData)))
|
||||
log.Log.Debug("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData)))
|
||||
}
|
||||
|
||||
deadline := time.Now().Add(75 * time.Millisecond)
|
||||
for {
|
||||
data := t.readAvailable()
|
||||
if len(data) > 0 {
|
||||
log.Log.Info("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data)))
|
||||
log.Log.Debug("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data)))
|
||||
return data, nil
|
||||
}
|
||||
|
||||
@@ -144,7 +144,7 @@ func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) {
|
||||
if stderr := t.stderrString(); stderr != "" {
|
||||
log.Log.Warning("webrtc.aac_transcoder: no output before deadline, ffmpeg stderr: " + stderr)
|
||||
} else {
|
||||
log.Log.Info("webrtc.aac_transcoder: no PCMU output before deadline")
|
||||
log.Log.Debug("webrtc.aac_transcoder: no PCMU output before deadline")
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
@@ -988,7 +988,7 @@ func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster
|
||||
if len(pcmu) == 0 {
|
||||
state.aacNoOutput++
|
||||
if state.aacNoOutput <= 5 || state.aacNoOutput%100 == 0 {
|
||||
log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data)))
|
||||
log.Log.Debug(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data)))
|
||||
}
|
||||
return // decoder still buffering
|
||||
}
|
||||
@@ -1004,7 +1004,7 @@ func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster
|
||||
state.lastAudioSample.Duration = sampleDuration(pkt, state.lastAudioSample.PacketTimestamp, 20*time.Millisecond)
|
||||
state.audioSamplesSent++
|
||||
if state.audioSamplesSent <= 5 || state.audioSamplesSent%100 == 0 {
|
||||
log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount()))
|
||||
log.Log.Debug(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount()))
|
||||
}
|
||||
audioBroadcaster.WriteSample(*state.lastAudioSample)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user