Implement remote resolver for storage indexer

CHANGELOG.md

@@ -32,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Prepare stub for Storage Indexer. Disable fetching packages from Package Storage v1. [#811](https://github.com/elastic/package-registry/pull/811)
 * Support input packages. [#809](https://github.com/elastic/package-registry/pull/809)
 * Implement storage indexer. [#814](https://github.com/elastic/package-registry/pull/814)
+* Implement hijack handlers for storage indexer. [#823](https://github.com/elastic/package-registry/pull/823)
 
 ### Deprecated
 

artifacts.go

@@ -8,6 +8,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/elastic/package-registry/storage"
+
 	"github.com/Masterminds/semver/v3"
 	"github.com/gorilla/mux"
 	"github.com/pkg/errors"
@@ -21,7 +23,7 @@ const artifactsRouterPath = "/epr/{packageName}/{packageName:[a-z0-9_]+}-{packag
 
 var errArtifactNotFound = errors.New("artifact not found")
 
-func artifactsHandler(indexer Indexer, cacheTime time.Duration) func(w http.ResponseWriter, r *http.Request) {
+func artifactsHandler(indexer packages.Indexer, cacheTime time.Duration) func(w http.ResponseWriter, r *http.Request) {
 	logger := util.Logger()
 	return func(w http.ResponseWriter, r *http.Request) {
 		vars := mux.Vars(r)
@@ -58,7 +60,13 @@ func artifactsHandler(indexer Indexer, cacheTime time.Duration) func(w http.Resp
 			return
 		}
 
+		aPackage := packageList[0]
+		if storageIndexer, assert := aPackage.Indexer().(*storage.Indexer); assert {
+			storageIndexer.HijackArtifactsHandler(w, r, aPackage)
+			return
+		}
+
 		cacheHeaders(w, cacheTime)
-		packages.ServePackage(w, r, packageList[0])
+		packages.ServeLocalPackage(w, r, packageList[0], packageList[0].BasePath)
 	}
 }

categories.go

@@ -27,7 +27,7 @@ type Category struct {
 }
 
 // categoriesHandler is a dynamic handler as it will also allow filtering in the future.
-func categoriesHandler(indexer Indexer, cacheTime time.Duration) func(w http.ResponseWriter, r *http.Request) {
+func categoriesHandler(indexer packages.Indexer, cacheTime time.Duration) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		query := r.URL.Query()
 

main.go

@@ -47,7 +47,7 @@ var (
 
 	featureStorageIndexer        bool
 	storageIndexerBucketInternal string
-	storageIndexerBucketPublic   string
+	storageEndpoint              string
 	storageIndexerWatchInterval  time.Duration
 
 	defaultConfig = Config{
@@ -70,7 +70,7 @@ func init() {
 	// The following storage related flags are technical preview and might be removed in the future or renamed
 	flag.BoolVar(&featureStorageIndexer, "feature-storage-indexer", false, "Enable storage indexer to include packages from Package Storage v2 (technical preview).")
 	flag.StringVar(&storageIndexerBucketInternal, "storage-indexer-bucket-internal", "", "Path to the internal Package Storage bucket (with gs:// prefix).")
-	flag.StringVar(&storageIndexerBucketPublic, "storage-indexer-bucket-public", "", "Path to the public Package Storage bucket (with gs:// prefix).")
+	flag.StringVar(&storageEndpoint, "storage-endpoint", "https://package-storage.elastic.co/", "Package Storage public endpoint.")
 	flag.DurationVar(&storageIndexerWatchInterval, "storage-indexer-watch-interval", 1*time.Minute, "Address of the package-registry service.")
 
 }
@@ -136,22 +136,22 @@ func initServer(logger *zap.Logger) *http.Server {
 	config := mustLoadConfig(logger)
 	packagesBasePaths := getPackagesBasePaths(config)
 
-	var indexers []Indexer
+	var indexers []packages.Indexer
 	if featureStorageIndexer {
 		storageClient, err := gstorage.NewClient(ctx)
 		if err != nil {
 			logger.Fatal("can't initialize storage client", zap.Error(err))
 		}
 		indexers = append(indexers, storage.NewIndexer(storageClient, storage.IndexerOptions{
 			PackageStorageBucketInternal: storageIndexerBucketInternal,
-			PackageStorageBucketPublic:   storageIndexerBucketPublic,
+			PackageStorageEndpoint:       storageEndpoint,
 			WatchInterval:                storageIndexerWatchInterval,
 		}))
 	} else {
 		indexers = append(indexers, packages.NewFileSystemIndexer(packagesBasePaths...))
 		indexers = append(indexers, packages.NewZipFileSystemIndexer(packagesBasePaths...))
 	}
-	combinedIndexer := NewCombinedIndexer(indexers...)
+	combinedIndexer := packages.NewCombinedIndexer(indexers...)
 	ensurePackagesAvailable(ctx, logger, combinedIndexer)
 
 	// If -dry-run=true is set, service stops here after validation
@@ -228,7 +228,7 @@ func printConfig(logger *zap.Logger, config *Config) {
 	logger.Info("Cache time for all others: " + config.CacheTimeCatchAll.String())
 }
 
-func ensurePackagesAvailable(ctx context.Context, logger *zap.Logger, indexer Indexer) {
+func ensurePackagesAvailable(ctx context.Context, logger *zap.Logger, indexer packages.Indexer) {
 	err := indexer.Init(ctx)
 	if err != nil {
 		logger.Fatal("Init failed", zap.Error(err))
@@ -246,15 +246,15 @@ func ensurePackagesAvailable(ctx context.Context, logger *zap.Logger, indexer In
 	logger.Info(fmt.Sprintf("%v package manifests loaded", len(packages)))
 }
 
-func mustLoadRouter(logger *zap.Logger, config *Config, indexer Indexer) *mux.Router {
+func mustLoadRouter(logger *zap.Logger, config *Config, indexer packages.Indexer) *mux.Router {
 	router, err := getRouter(logger, config, indexer)
 	if err != nil {
 		logger.Fatal("failed go configure router", zap.Error(err))
 	}
 	return router
 }
 
-func getRouter(logger *zap.Logger, config *Config, indexer Indexer) (*mux.Router, error) {
+func getRouter(logger *zap.Logger, config *Config, indexer packages.Indexer) (*mux.Router, error) {
 	artifactsHandler := artifactsHandler(indexer, config.CacheTimeCatchAll)
 	signaturesHandler := signaturesHandler(indexer, config.CacheTimeCatchAll)
 	faviconHandleFunc, err := faviconHandler(config.CacheTimeCatchAll)

main_test.go

@@ -245,7 +245,7 @@ func TestStaticsModifiedTime(t *testing.T) {
 		},
 	}
 
-	indexer := NewCombinedIndexer(
+	indexer := packages.NewCombinedIndexer(
 		packages.NewFileSystemIndexer("./testdata/package"),
 		packages.NewZipFileSystemIndexer("./testdata/local-storage"),
 	)
@@ -306,7 +306,7 @@ func TestZippedArtifacts(t *testing.T) {
 }
 
 func TestPackageIndex(t *testing.T) {
-	indexer := NewCombinedIndexer(
+	indexer := packages.NewCombinedIndexer(
 		packages.NewFileSystemIndexer("./testdata/package"),
 		packages.NewZipFileSystemIndexer("./testdata/local-storage"),
 	)
@@ -424,7 +424,7 @@ func TestContentTypes(t *testing.T) {
 		{"/package/example/1.0.1/img/kibana-envoyproxy.jpg", "image/jpeg"},
 	}
 
-	indexer := NewCombinedIndexer(
+	indexer := packages.NewCombinedIndexer(
 		packages.NewFileSystemIndexer("./testdata/package"),
 		packages.NewZipFileSystemIndexer("./testdata/local-storage"),
 	)
@@ -453,7 +453,7 @@ func TestContentTypes(t *testing.T) {
 // TestRangeDownloads tests that range downloads continue working for packages stored
 // on different file systems.
 func TestRangeDownloads(t *testing.T) {
-	indexer := NewCombinedIndexer(
+	indexer := packages.NewCombinedIndexer(
 		packages.NewFileSystemIndexer("./testdata/package"),
 		packages.NewZipFileSystemIndexer("./testdata/local-storage"),
 	)

package_index.go

@@ -24,7 +24,7 @@ const (
 
 var errPackageRevisionNotFound = errors.New("package revision not found")
 
-func packageIndexHandler(indexer Indexer, cacheTime time.Duration) func(w http.ResponseWriter, r *http.Request) {
+func packageIndexHandler(indexer packages.Indexer, cacheTime time.Duration) func(w http.ResponseWriter, r *http.Request) {
 	logger := util.Logger()
 	return func(w http.ResponseWriter, r *http.Request) {
 		vars := mux.Vars(r)

package_storage_test.go

@@ -0,0 +1,183 @@
+// 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 main
+
+import (
+	"context"
+	"net/http"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/package-registry/storage"
+)
+
+const storageIndexerGoldenDir = "storage-indexer"
+
+func TestPackageStorage_Endpoints(t *testing.T) {
+	fs := storage.PrepareFakeServer(t, "./storage/testdata/search-index-all-full.json")
+	defer fs.Stop()
+	indexer := storage.NewIndexer(fs.Client(), storage.FakeIndexerOptions)
+
+	err := indexer.Init(context.Background())
+	require.NoError(t, err)
+
+	tests := []struct {
+		endpoint string
+		path     string
+		file     string
+		handler  func(w http.ResponseWriter, r *http.Request)
+	}{
+		{"/search", "/search", "search.json", searchHandler(indexer, testCacheTime)},
+		{"/search?all=true", "/search", "search-all.json", searchHandler(indexer, testCacheTime)},
+		{"/categories", "/categories", "categories.json", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?experimental=true", "/categories", "categories-experimental.json", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?experimental=foo", "/categories", "categories-experimental-error.txt", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?experimental=true&kibana.version=6.5.2", "/categories", "categories-kibana652.json", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?prerelease=true", "/categories", "categories-prerelease.json", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?prerelease=foo", "/categories", "categories-prerelease-error.txt", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?prerelease=true&kibana.version=6.5.2", "/categories", "categories-prerelease-kibana652.json", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?include_policy_templates=true", "/categories", "categories-include-policy-templates.json", categoriesHandler(indexer, testCacheTime)},
+		{"/categories?include_policy_templates=foo", "/categories", "categories-include-policy-templates-error.txt", categoriesHandler(indexer, testCacheTime)},
+		{"/search?kibana.version=6.5.2", "/search", "search-kibana652.json", searchHandler(indexer, testCacheTime)},
+		{"/search?kibana.version=7.2.1", "/search", "search-kibana721.json", searchHandler(indexer, testCacheTime)},
+		{"/search?kibana.version=8.0.0", "/search", "search-kibana800.json", searchHandler(indexer, testCacheTime)},
+		{"/search?category=web", "/search", "search-category-web.json", searchHandler(indexer, testCacheTime)},
+		{"/search?category=web&all=true", "/search", "search-category-web-all.json", searchHandler(indexer, testCacheTime)},
+		{"/search?category=custom", "/search", "search-category-custom.json", searchHandler(indexer, testCacheTime)},
+		{"/search?experimental=true", "/search", "search-package-experimental.json", searchHandler(indexer, testCacheTime)},
+		{"/search?experimental=foo", "/search", "search-package-experimental-error.txt", searchHandler(indexer, testCacheTime)},
+		{"/search?category=datastore&experimental=true", "/search", "search-category-datastore.json", searchHandler(indexer, testCacheTime)},
+		{"/search?prerelease=true", "/search", "search-package-prerelease.json", searchHandler(indexer, testCacheTime)},
+		{"/search?prerelease=foo", "/search", "search-package-prerelease-error.txt", searchHandler(indexer, testCacheTime)},
+		{"/search?category=datastore&prerelease=true", "/search", "search-category-datastore-prerelease.json", searchHandler(indexer, testCacheTime)},
+
+		// Removed flags, kept ensure that they don't break requests from old versions.
+		{"/search?internal=true", "/search", "search-package-internal.json", searchHandler(indexer, testCacheTime)},
+	}
+
+	for _, test := range tests {
+		t.Run(test.endpoint, func(t *testing.T) {
+			runEndpointWithStorageIndexer(t, test.endpoint, test.path, test.file, test.handler)
+		})
+	}
+}
+
+func TestPackageStorage_PackageIndex(t *testing.T) {
+	fs := storage.PrepareFakeServer(t, "./storage/testdata/search-index-all-full.json")
+	defer fs.Stop()
+	indexer := storage.NewIndexer(fs.Client(), storage.FakeIndexerOptions)
+
+	err := indexer.Init(context.Background())
+	require.NoError(t, err)
+
+	packageIndexHandler := packageIndexHandler(indexer, testCacheTime)
+
+	tests := []struct {
+		endpoint string
+		path     string
+		file     string
+		handler  func(w http.ResponseWriter, r *http.Request)
+	}{
+		{"/package/1password/0.1.1/", packageIndexRouterPath, "1password-0.1.1.json", packageIndexHandler},
+		{"/package/kubernetes/0.3.0/", packageIndexRouterPath, "kubernetes-0.3.0.json", packageIndexHandler},
+		{"/package/osquery/1.0.3/", packageIndexRouterPath, "osquery-1.0.3.json", packageIndexHandler},
+	}
+
+	for _, test := range tests {
+		t.Run(test.endpoint, func(t *testing.T) {
+			runEndpointWithStorageIndexer(t, test.endpoint, test.path, test.file, test.handler)
+		})
+	}
+}
+
+func TestPackageStorage_Artifacts(t *testing.T) {
+	fs := storage.PrepareFakeServer(t, "./storage/testdata/search-index-all-full.json")
+	defer fs.Stop()
+	indexer := storage.NewIndexer(fs.Client(), storage.FakeIndexerOptions)
+
+	err := indexer.Init(context.Background())
+	require.NoError(t, err)
+
+	artifactsHandler := artifactsHandler(indexer, testCacheTime)
+
+	tests := []struct {
+		endpoint string
+		path     string
+		file     string
+		handler  func(w http.ResponseWriter, r *http.Request)
+	}{
+		{"/epr/1password/1password-0.1.1.zip", artifactsRouterPath, "1password-0.1.1.zip.txt", artifactsHandler},
+		{"/epr/kubernetes/kubernetes-999.999.999.zip", artifactsRouterPath, "artifact-package-version-not-found.txt", artifactsHandler},
+		{"/epr/missing/missing-1.0.3.zip", artifactsRouterPath, "artifact-package-not-found.txt", artifactsHandler},
+	}
+
+	for _, test := range tests {
+		t.Run(test.endpoint, func(t *testing.T) {
+			runEndpointWithStorageIndexer(t, test.endpoint, test.path, test.file, test.handler)
+		})
+	}
+}
+
+func TestPackageStorage_Signatures(t *testing.T) {
+	fs := storage.PrepareFakeServer(t, "./storage/testdata/search-index-all-full.json")
+	defer fs.Stop()
+	indexer := storage.NewIndexer(fs.Client(), storage.FakeIndexerOptions)
+
+	err := indexer.Init(context.Background())
+	require.NoError(t, err)
+
+	signaturesHandler := signaturesHandler(indexer, testCacheTime)
+
+	tests := []struct {
+		endpoint string
+		path     string
+		file     string
+		handler  func(w http.ResponseWriter, r *http.Request)
+	}{
+		{"/epr/1password/1password-0.1.1.zip.sig", signaturesRouterPath, "1password-0.1.1.zip.sig", signaturesHandler},
+		{"/epr/checkpoint/checkpoint-0.5.2.zip.sig", signaturesRouterPath, "checkpoint-0.5.2.zip.sig", signaturesHandler},
+	}
+
+	for _, test := range tests {
+		t.Run(test.endpoint, func(t *testing.T) {
+			runEndpoint(t, test.endpoint, test.path, test.file, test.handler)
+		})
+	}
+}
+
+func TestPackageStorage_Statics(t *testing.T) {
+	fs := storage.PrepareFakeServer(t, "./storage/testdata/search-index-all-full.json")
+	defer fs.Stop()
+	indexer := storage.NewIndexer(fs.Client(), storage.FakeIndexerOptions)
+
+	err := indexer.Init(context.Background())
+	require.NoError(t, err)
+
+	staticHandler := staticHandler(indexer, testCacheTime)
+
+	tests := []struct {
+		endpoint string
+		path     string
+		file     string
+		handler  func(w http.ResponseWriter, r *http.Request)
+	}{
+		{"/package/1password/0.1.1/img/1password-logo-light-bg.svg", staticRouterPath, "1password-logo-light-bg.svg", staticHandler},
+		{"/package/cassandra/1.1.0/img/[Logs Cassandra] System Logs.jpg", staticRouterPath, "logs-cassandra-system-logs.jpg", staticHandler},
+		{"/package/cef/0.1.0/docs/README.md", staticRouterPath, "cef-readme.md", staticHandler},
+	}
+
+	for _, test := range tests {
+		t.Run(test.endpoint, func(t *testing.T) {
+			runEndpoint(t, test.endpoint, test.path, test.file, test.handler)
+		})
+	}
+
+}
+
+func runEndpointWithStorageIndexer(t *testing.T, endpoint, path, file string, handler func(w http.ResponseWriter, r *http.Request)) {
+	runEndpoint(t, endpoint, path, filepath.Join(storageIndexerGoldenDir, file), handler)
+}