File delivery

NOTICE.txt

@@ -1513,6 +1513,36 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/go-ucfg@v0.8.6/
    limitations under the License.
 
 
+--------------------------------------------------------------------------------
+Dependency : github.com/fxamacker/cbor/v2
+Version: v2.4.0
+Licence type (autodetected): MIT
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/fxamacker/cbor/v2@v2.4.0/LICENSE:
+
+MIT License
+
+Copyright (c) 2019-present Faye Amacker
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
 --------------------------------------------------------------------------------
 Dependency : github.com/go-chi/chi/v5
 Version: v5.0.8
@@ -42259,6 +42289,38 @@ SOFTWARE.
 
 
 
+--------------------------------------------------------------------------------
+Dependency : github.com/x448/float16
+Version: v0.8.4
+Licence type (autodetected): MIT
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/x448/float16@v0.8.4/LICENSE:
+
+MIT License
+
+Copyright (c) 2019 Montgomery Edwards⁴⁴⁸ and Faye Amacker
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
+
+
 --------------------------------------------------------------------------------
 Dependency : github.com/xhit/go-str2duration
 Version: v1.2.0

changelog/fragments/1685985303-file-delivery.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: File Delivery to integrations
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; a word indicating the component this changeset affects.
+component:
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: 2666
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/elastic/elastic-agent-system-metrics v0.6.1
 	github.com/elastic/go-elasticsearch/v8 v8.7.0
 	github.com/elastic/go-ucfg v0.8.6
+	github.com/fxamacker/cbor/v2 v2.4.0
 	github.com/go-chi/chi/v5 v5.0.8
 	github.com/gofrs/uuid v4.3.1+incompatible
 	github.com/google/go-cmp v0.5.9
@@ -79,6 +80,7 @@ require (
 	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.11 // indirect
 	github.com/tklauser/numcpus v0.6.0 // indirect
+	github.com/x448/float16 v0.8.4 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.elastic.co/ecszap v1.0.1 // indirect
 	go.elastic.co/fastjson v1.1.0 // indirect

go.sum

@@ -204,6 +204,8 @@ github.com/envoyproxy/go-control-plane v0.10.3 h1:xdCVXxEe0Y3FQith+0cj2irwZudqGY
 github.com/envoyproxy/protoc-gen-validate v0.9.1 h1:PS7VIOgmSVhWUEeZwTe7z7zouA22Cr590PzXKbZHOVY=
 github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
+github.com/fxamacker/cbor/v2 v2.4.0 h1:ri0ArlOR+5XunOP8CRUowT0pSJOwhW098ZCUyskZD88=
+github.com/fxamacker/cbor/v2 v2.4.0/go.mod h1:TA1xS00nchWmaBnEIxPSE5oHLuJBAVvqrtAnWBwBCVo=
 github.com/getkin/kin-openapi v0.107.0 h1:bxhL6QArW7BXQj8NjXfIJQy680NsMKd25nwhvpCXchg=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-gonic/gin v1.8.2 h1:UzKToD9/PoFj/V4rvlKqTRKnQYyz8Sc1MJlv4JHPtvY=
@@ -448,6 +450,8 @@ github.com/tklauser/numcpus v0.6.0/go.mod h1:FEZLMke0lhOUG6w2JadTzp0a+Nl8PF/GFkQ
 github.com/ugorji/go/codec v1.2.9 h1:rmenucSohSTiyL09Y+l2OCk+FrMxGMzho2+tjr5ticU=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo=
+github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
+github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xhit/go-str2duration v1.2.0 h1:BcV5u025cITWxEQKGWr1URRzrcXtu7uk8+luz3Yuhwc=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

internal/pkg/api/api.go

@@ -26,6 +26,7 @@ type apiServer struct {
 	sm     policy.SelfMonitor
 	bi     build.Info
 	ut     *UploadT
+	ft     *FileDeliveryT
 	bulker bulk.Bulk
 }
 
@@ -128,6 +129,14 @@ func (a *apiServer) UploadChunk(w http.ResponseWriter, r *http.Request, id strin
 	}
 }
 
+func (a *apiServer) GetFile(w http.ResponseWriter, r *http.Request, id string) {
+	zlog := hlog.FromRequest(r).With().Logger()
+	if err := a.ft.handleSendFile(zlog, w, r, id); err != nil {
+		cntFileDeliv.IncError(err)
+		ErrorResp(w, r, err)
+	}
+}
+
 func (a *apiServer) Status(w http.ResponseWriter, r *http.Request, params StatusParams) {
 	zlog := hlog.FromRequest(r).With().
 		Str("mod", kStatusMod).

internal/pkg/api/error.go

@@ -15,6 +15,8 @@ import (
 
 	"github.com/elastic/fleet-server/v7/internal/pkg/apikey"
 	"github.com/elastic/fleet-server/v7/internal/pkg/dl"
+	"github.com/elastic/fleet-server/v7/internal/pkg/file"
+	"github.com/elastic/fleet-server/v7/internal/pkg/file/delivery"
 	"github.com/elastic/fleet-server/v7/internal/pkg/file/uploader"
 	"github.com/elastic/fleet-server/v7/internal/pkg/limit"
 	"github.com/elastic/fleet-server/v7/internal/pkg/logger"
@@ -386,6 +388,25 @@ func NewHTTPErrResp(err error) HTTPErrResp {
 				zerolog.InfoLevel,
 			},
 		},
+		// file
+		{
+			delivery.ErrNoFile,
+			HTTPErrResp{
+				http.StatusNotFound,
+				"ErrNoFile",
+				"file not found",
+				zerolog.InfoLevel,
+			},
+		},
+		{
+			file.ErrInvalidID,
+			HTTPErrResp{
+				http.StatusBadRequest,
+				"ErrInvalidFileID",
+				"ErrInvalidID",
+				zerolog.InfoLevel,
+			},
+		},
 	}
 
 	for _, e := range errTable {

internal/pkg/api/handleFileDelivery.go

@@ -0,0 +1,77 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/elastic/fleet-server/v7/internal/pkg/apikey"
+	"github.com/elastic/fleet-server/v7/internal/pkg/bulk"
+	"github.com/elastic/fleet-server/v7/internal/pkg/cache"
+	"github.com/elastic/fleet-server/v7/internal/pkg/config"
+	"github.com/elastic/fleet-server/v7/internal/pkg/file/delivery"
+	"github.com/elastic/fleet-server/v7/internal/pkg/model"
+	"github.com/elastic/go-elasticsearch/v8"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+type FileDeliveryT struct {
+	bulker      bulk.Bulk
+	cache       cache.Cache
+	chunkClient *elasticsearch.Client
+	deliverer   *delivery.Deliverer
+	authAgent   func(*http.Request, *string, bulk.Bulk, cache.Cache) (*model.Agent, error) // injectable for testing purposes
+	authAPIKey  func(*http.Request, bulk.Bulk, cache.Cache) (*apikey.APIKey, error)        // as above
+}
+
+func NewFileDeliveryT(cfg *config.Server, bulker bulk.Bulk, chunkClient *elasticsearch.Client, cache cache.Cache) *FileDeliveryT {
+	log.Info().
+		Interface("limits", cfg.Limits.ArtifactLimit).
+		Int64("maxFileSize", maxFileSize).
+		Msg("upload limits")
+
+	return &FileDeliveryT{
+		chunkClient: chunkClient,
+		bulker:      bulker,
+		cache:       cache,
+		deliverer:   delivery.New(chunkClient, bulker, maxFileSize),
+		authAgent:   authAgent,
+		authAPIKey:  authAPIKey,
+	}
+}
+
+func (ft *FileDeliveryT) handleSendFile(zlog zerolog.Logger, w http.ResponseWriter, r *http.Request, fileID string) error {
+	agent, err := authAgent(r, nil, ft.bulker, ft.cache)
+	if err != nil {
+		return err
+	}
+
+	// find file
+	info, err := ft.deliverer.FindFileForAgent(r.Context(), fileID, agent.Agent.ID)
+	if err != nil {
+		return err
+	}
+
+	// set headers before writing any chunks!
+
+	// if mime_type was provided, set as Content-Type, otherwise fall back to octet-stream
+	w.Header().Set("Content-Type", "application/octet-stream")
+	if info.File.MimeType != "" {
+		w.Header().Set("Content-Type", info.File.MimeType)
+	}
+
+	if info.File.Size > 0 {
+		w.Header().Set("Content-Length", strconv.FormatInt(info.File.Size, 10))
+	}
+
+	if info.File.Hash != nil && info.File.Hash.SHA2 != "" {
+		w.Header().Set("X-File-SHA2", info.File.Hash.SHA2)
+	}
+
+	// stream the chunks out
+	return ft.deliverer.SendFile(r.Context(), zlog, w, info, fileID)
+}

internal/pkg/api/metrics.go

@@ -42,6 +42,7 @@ var (
 	cntUploadStart routeStats
 	cntUploadChunk routeStats
 	cntUploadEnd   routeStats
+	cntFileDeliv   routeStats
 	cntArtifacts   artifactStats
 
 	infoReg sync.Once
@@ -71,6 +72,8 @@ func init() {
 	cntUploadStart.Register(routesRegistry.newRegistry("uploadStart"))
 	cntUploadChunk.Register(routesRegistry.newRegistry("uploadChunk"))
 	cntUploadEnd.Register(routesRegistry.newRegistry("uploadEnd"))
+	cntFileDeliv.Register(routesRegistry.newRegistry("deliverFile"))
+
 }
 
 // metricsRegistry wraps libbeat and prometheus registries