image api

api/http_internal.go

@@ -23,6 +23,7 @@ import (
 	"github.com/livepeer/catalyst-api/handlers/misttriggers"
 	"github.com/livepeer/catalyst-api/log"
 	mistapiconnector "github.com/livepeer/catalyst-api/mapic"
+	"github.com/livepeer/catalyst-api/metrics"
 	"github.com/livepeer/catalyst-api/middleware"
 	"github.com/livepeer/catalyst-api/pipeline"
 	"github.com/livepeer/go-api-client"
@@ -113,6 +114,14 @@ func NewCatalystAPIRouterInternal(cli config.Cli, vodEngine *pipeline.Coordinato
 			),
 		),
 	)
+	router.GET("/api/image/:playbackID/thumbnail.jpg",
+		middleware.LogAndMetrics(metrics.Metrics.ImageAPIDurationSec)(
+			withAuth(
+				cli.APIToken,
+				handlers.NewImageHandler(cli.PrivateBucketURLs).Handle,
+			),
+		),
+	)
 
 	// Public handler to propagate an event to all Catalyst nodes
 	router.POST("/api/events", withLogging(eventsHandler.Events()))

config/cli.go

@@ -41,6 +41,7 @@ type Cli struct {
 	APIServer                 string
 	SourceOutput              string
 	PrivateBucketURLs         []*url.URL
+	PublicBucketURLs          []*url.URL
 	ExternalTranscoder        string
 	VodPipelineStrategy       string
 	MetricsDBConnectionString string

handlers/image.go

@@ -0,0 +1,199 @@
+package handlers
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	"github.com/grafov/m3u8"
+	"github.com/julienschmidt/httprouter"
+	caterrs "github.com/livepeer/catalyst-api/errors"
+	"github.com/livepeer/catalyst-api/log"
+	"github.com/livepeer/catalyst-api/metrics"
+	"github.com/livepeer/catalyst-api/playback"
+	ffmpeg "github.com/u2takey/ffmpeg-go"
+)
+
+type ImageHandler struct {
+	PublicBucketURLs []*url.URL
+}
+
+func NewImageHandler(urls []*url.URL) *ImageHandler {
+	return &ImageHandler{
+		PublicBucketURLs: urls,
+	}
+}
+
+func (p *ImageHandler) Handle(w http.ResponseWriter, req *http.Request, params httprouter.Params) {
+	timeString := req.URL.Query().Get("time")
+	time, err := strconv.ParseFloat(timeString, 64)
+	if err != nil {
+		log.LogNoRequestID("image API error", "err", err)
+		caterrs.WriteHTTPBadRequest(w, "failed to parse time", nil)
+		return
+	}
+	width, err := parseResolution(req, "width", 320)
+	if err != nil {
+		log.LogNoRequestID("image API error", "err", err)
+		caterrs.WriteHTTPBadRequest(w, "failed to parse width", nil)
+		return
+	}
+	height, err := parseResolution(req, "height", 240)
+	if err != nil {
+		log.LogNoRequestID("image API error", "err", err)
+		caterrs.WriteHTTPBadRequest(w, "failed to parse height", nil)
+		return
+	}
+
+	playbackID := params.ByName("playbackID")
+	if playbackID == "" {
+		caterrs.WriteHTTPBadRequest(w, "playbackID was empty", nil)
+		return
+	}
+
+	err = p.handle(w, playbackID, time, width, height)
+	if err != nil {
+		log.LogNoRequestID("image API error", "err", err)
+		switch {
+		case errors.Is(err, caterrs.ObjectNotFoundError):
+			caterrs.WriteHTTPNotFound(w, "not found", nil)
+		default:
+			caterrs.WriteHTTPInternalServerError(w, "internal server error", nil)
+		}
+	}
+}
+
+func parseResolution(req *http.Request, key string, defaultVal int64) (int64, error) {
+	val := req.URL.Query().Get(key)
+	if val == "" {
+		return defaultVal, nil
+	}
+	return strconv.ParseInt(val, 10, 32)
+}
+
+func (p *ImageHandler) handle(w http.ResponseWriter, playbackID string, t float64, width int64, height int64) error {
+	var (
+		err         error
+		segmentFile string
+		dir         = playbackID
+		start       = time.Now()
+	)
+
+	// download master playlist
+	fileInfoReader, err := playback.OsFetch(p.PublicBucketURLs, dir, "index.m3u8", "")
+	if err != nil {
+		return fmt.Errorf("failed to read manifest: %w", err)
+	}
+
+	manifest, _, err := m3u8.DecodeFrom(fileInfoReader.Body, true)
+	if err != nil {
+		return fmt.Errorf("failed to decode manifest: %w", err)
+	}
+	masterPlaylist, ok := manifest.(*m3u8.MasterPlaylist)
+	if !ok || masterPlaylist == nil {
+		return fmt.Errorf("failed to parse playlist as MasterPlaylist")
+	}
+	if len(masterPlaylist.Variants) < 1 {
+		return fmt.Errorf("no renditions found")
+	}
+
+	// download rendition playlist
+	renditionUri := masterPlaylist.Variants[0].URI
+	fileInfoReader, err = playback.OsFetch(p.PublicBucketURLs, dir, renditionUri, "")
+	if err != nil {
+		return fmt.Errorf("failed to read manifest: %w", err)
+	}
+	dir = filepath.Join(dir, filepath.Dir(renditionUri))
+	manifest, _, err = m3u8.DecodeFrom(fileInfoReader.Body, true)
+	if err != nil {
+		return fmt.Errorf("failed to decode manifest: %w", err)
+	}
+	mediaPlaylist, ok := manifest.(*m3u8.MediaPlaylist)
+	if !ok || mediaPlaylist == nil {
+		return fmt.Errorf("failed to parse playlist as MediaPlaylist")
+	}
+
+	// find the segment required
+	currentTime := 0.0
+	extractTime := 0.0
+	for _, segment := range mediaPlaylist.GetAllSegments() {
+		currentTime += segment.Duration
+		if currentTime > t {
+			segmentFile = segment.URI
+			extractTime = t - currentTime + segment.Duration
+			break
+		}
+	}
+	if segmentFile == "" {
+		return fmt.Errorf("playbackID media not found: %w", caterrs.ObjectNotFoundError)
+	}
+
+	tmpDir, err := os.MkdirTemp(os.TempDir(), "image-api-*")
+	if err != nil {
+		return fmt.Errorf("temp file creation failed: %w", err)
+	}
+	defer os.RemoveAll(tmpDir)
+
+	// download the segment
+	fileInfoReader, err = playback.OsFetch(p.PublicBucketURLs, dir, segmentFile, "")
+	if err != nil {
+		return fmt.Errorf("failed to get media: %w", err)
+	}
+	segBytes, err := io.ReadAll(fileInfoReader.Body)
+	if err != nil {
+		return fmt.Errorf("failed to get bytes: %w", err)
+	}
+
+	inputFile := path.Join(tmpDir, "in.ts")
+	if err = os.WriteFile(inputFile, segBytes, 0644); err != nil {
+		return fmt.Errorf("failed to write input file: %w", err)
+	}
+	outputFile := path.Join(tmpDir, "out.jpg")
+
+	metrics.Metrics.ImageAPIDownloadDurationSec.WithLabelValues().Observe(time.Since(start).Seconds())
+
+	// extract image
+	extractStart := time.Now()
+	defer func() {
+		metrics.Metrics.ImageAPIExtractDurationSec.WithLabelValues().Observe(time.Since(extractStart).Seconds())
+	}()
+
+	var ffmpegErr bytes.Buffer
+	err = ffmpeg.
+		Input(inputFile).
+		Output(
+			outputFile,
+			ffmpeg.KwArgs{
+				"ss":      fmt.Sprintf("00:00:%d", int64(extractTime)),
+				"vframes": "1",
+				"vf":      fmt.Sprintf("scale=%d:%d:force_original_aspect_ratio=decrease", width, height),
+			},
+		).OverWriteOutput().WithErrorOutput(&ffmpegErr).Run()
+	if err != nil {
+		return fmt.Errorf("ffmpeg failed [%s]: %w", ffmpegErr.String(), err)
+	}
+
+	bs, err := os.ReadFile(outputFile)
+	if err != nil {
+		return err
+	}
+
+	w.Header().Set("content-type", "image/jpg")
+	w.Header().Set("content-length", strconv.Itoa(len(bs)))
+	w.WriteHeader(http.StatusOK)
+	count, err := w.Write(bs)
+	if err != nil {
+		log.LogNoRequestID("image handler failed to write response", "err", err)
+	} else {
+		log.LogNoRequestID("image handler wrote response", "count", count)
+	}
+	return nil
+}

handlers/image_test.go

@@ -0,0 +1,98 @@
+package handlers
+
+import (
+	"context"
+	"io"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"testing"
+
+	"github.com/julienschmidt/httprouter"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/vansante/go-ffprobe.v2"
+)
+
+func TestImageHandler_Handle(t *testing.T) {
+	wd, err := os.Getwd()
+	require.NoError(t, err)
+	handler := &ImageHandler{
+		PublicBucketURLs: []*url.URL{{Scheme: "file", Path: wd + "/../test"}},
+	}
+	tests := []struct {
+		name           string
+		time           string
+		playbackID     string
+		expectedStatus int
+	}{
+		{
+			name: "first segment",
+			time: "5",
+		},
+		{
+			name: "second segment",
+			time: "21",
+		},
+		{
+			name: "final segment",
+			time: "29",
+		},
+		{
+			name:           "out of bounds",
+			time:           "30",
+			expectedStatus: http.StatusNotFound,
+		},
+		{
+			name:           "invalid time",
+			time:           "",
+			expectedStatus: http.StatusBadRequest,
+		},
+		{
+			name:           "playbackID not found",
+			time:           "29",
+			playbackID:     "foo",
+			expectedStatus: http.StatusNotFound,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			w := httptest.NewRecorder()
+			req, err := http.NewRequest(http.MethodGet, "?time="+tt.time, nil)
+			require.NoError(t, err)
+
+			if tt.playbackID == "" {
+				tt.playbackID = "fixtures" // just use the fixtures directory for testing
+			}
+			handler.Handle(w, req, []httprouter.Param{{
+				Key:   "playbackID",
+				Value: tt.playbackID,
+			}})
+			resp := w.Result()
+			if tt.expectedStatus == 0 {
+				tt.expectedStatus = 200
+			}
+			require.Equal(t, tt.expectedStatus, resp.StatusCode)
+
+			if tt.expectedStatus != 200 {
+				return
+			}
+			respBytes, err := io.ReadAll(resp.Body)
+			require.NoError(t, err)
+
+			outfile, err := os.CreateTemp(os.TempDir(), "out*.jpg")
+			require.NoError(t, err)
+			defer os.Remove(outfile.Name())
+			_, err = outfile.Write(respBytes)
+			require.NoError(t, err)
+			log.Println(outfile.Name())
+			probeData, err := ffprobe.ProbeURL(context.Background(), outfile.Name())
+			require.NoError(t, err)
+			require.Equal(t, "image2", probeData.Format.FormatName)
+			require.Len(t, probeData.Streams, 1)
+			require.Greater(t, probeData.Streams[0].Width, 0)
+			require.Greater(t, probeData.Streams[0].Height, 0)
+		})
+	}
+}

main.go

@@ -60,6 +60,7 @@ func main() {
 	fs.StringVar(&cli.APIToken, "api-token", "IAmAuthorized", "Auth header value for API access")
 	fs.StringVar(&cli.SourceOutput, "source-output", "", "URL for the video source segments used if source_segments is not defined in the upload request")
 	config.URLSliceVarFlag(fs, &cli.PrivateBucketURLs, "private-bucket", "", "URL for the private media bucket")
+	config.URLSliceVarFlag(fs, &cli.PublicBucketURLs, "public-bucket", "", "URL for the public media bucket")
 	fs.StringVar(&cli.ExternalTranscoder, "external-transcoder", "", "URL for the external transcoder to be used by the pipeline coordinator. Only 1 implementation today for AWS MediaConvert which should be in the format: mediaconvert://key-id:key-secret@endpoint-host?region=aws-region&role=iam-role&s3_aux_bucket=s3://bucket")
 	fs.StringVar(&cli.VodPipelineStrategy, "vod-pipeline-strategy", string(pipeline.StrategyCatalystFfmpegDominance), "Which strategy to use for the VOD pipeline")
 	fs.StringVar(&cli.MetricsDBConnectionString, "metrics-db-connection-string", "", "Connection string to use for the metrics Postgres DB. Takes the form: host=X port=X user=X password=X dbname=X")

metrics/metrics.go

@@ -34,6 +34,9 @@ type CatalystAPIMetrics struct {
 	SerfEventBufferSize             prometheus.Gauge
 	AccessControlRequestCount       *prometheus.CounterVec
 	AccessControlRequestDurationSec *prometheus.SummaryVec
+	ImageAPIDurationSec             *prometheus.SummaryVec
+	ImageAPIDownloadDurationSec     *prometheus.SummaryVec
+	ImageAPIExtractDurationSec      *prometheus.SummaryVec
 
 	JobsInFlight         prometheus.Gauge
 	HTTPRequestsInFlight prometheus.Gauge
@@ -112,6 +115,9 @@ func NewMetrics() *CatalystAPIMetrics {
 			Name: "access_control_request_duration_seconds",
 			Help: "The latency of the access control requests",
 		}, []string{"allowed", "playbackID"}),
+		ImageAPIDurationSec:         durationSummary("image_api_response_duration", "Total time taken to process Image API request", "success", "status_code", "version"),
+		ImageAPIDownloadDurationSec: durationSummary("image_api_download_duration", "Time taken to download media from storage while generating an image"),
+		ImageAPIExtractDurationSec:  durationSummary("image_api_extract_duration", "Time taken to generate image"),
 
 		// Clients metrics
 		TranscodingStatusUpdate: ClientMetrics{
@@ -212,4 +218,12 @@ func NewMetrics() *CatalystAPIMetrics {
 	return m
 }
 
+func durationSummary(name, help string, labelNames ...string) *prometheus.SummaryVec {
+	return promauto.NewSummaryVec(prometheus.SummaryOpts{
+		Name:       name,
+		Help:       help,
+		Objectives: map[float64]float64{0.5: 0.05, 0.8: 0.01, 0.95: 0.01},
+	}, labelNames)
+}
+
 var Metrics = NewMetrics()

playback/playback.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"net/url"
+	"path/filepath"
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws/awserr"
@@ -103,6 +104,10 @@ func appendAccessKey(uri, gatingParam, gatingParamName string) (string, error) {
 }
 
 func osFetch(buckets []*url.URL, playbackID, file, byteRange string) (*drivers.FileInfoReader, error) {
+	return OsFetch(buckets, filepath.Join("hls", playbackID), file, byteRange)
+}
+
+func OsFetch(buckets []*url.URL, playbackID, file, byteRange string) (*drivers.FileInfoReader, error) {
 	if len(buckets) < 1 {
 		return nil, errors.New("playback failed, no private buckets configured")
 	}
@@ -113,7 +118,7 @@ func osFetch(buckets []*url.URL, playbackID, file, byteRange string) (*drivers.F
 	)
 	// try all private buckets until object is found or return error
 	for _, bucket := range buckets {
-		osURL := bucket.JoinPath("hls").JoinPath(playbackID).JoinPath(file)
+		osURL := bucket.JoinPath(playbackID).JoinPath(file)
 		f, err = clients.GetOSURL(osURL.String(), byteRange)
 		if err == nil {
 			// object found successfully so return early