Return single package when multiple indexers are used

CHANGELOG.md

@@ -10,6 +10,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bugfixes
 
+* Return only the latest version of each package when a combined index is used. [#849](https://github.com/elastic/package-registry/pull/849)
+* Return only first appearance of the same version of a package when it is available in multiple indexes. [#849](https://github.com/elastic/package-registry/pull/849)
+
 ### Added
 
 * Add `elastic.subscription` condition to package index metadata, use this value for backwards compatibility with previous `license` field. [#826](https://github.com/elastic/package-registry/pull/826)

indexer.go

@@ -6,8 +6,11 @@ package main
 
 import (
  "context"
+ "sort"
  "time"
 
+ "github.com/Masterminds/semver/v3"
+
  "github.com/elastic/package-registry/metrics"
  "github.com/elastic/package-registry/packages"
 )
@@ -44,5 +47,44 @@ func (c CombinedIndexer) Get(ctx context.Context, opts *packages.GetOptions) (pa
  }
  packages = packages.Join(p)
  }
+
+ if !opts.Filter.AllVersions {
+ return latestPackagesVersion(packages), nil
+ }
+
  return packages, nil
 }
+
+func latestPackagesVersion(source packages.Packages) (result packages.Packages) {
+ sort.Sort(byNameVersion(source))
+
+ current := ""
+ for _, p := range source {
+ if p.Name == current {
+ continue
+ }
+
+ current = p.Name
+ result = append(result, p)
+ }
+
+ return result
+}
+
+type byNameVersion packages.Packages
+
+func (p byNameVersion) Len() int { return len(p) }
+func (p byNameVersion) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
+func (p byNameVersion) Less(i, j int) bool {
+ if p[i].Name != p[j].Name {
+ return p[i].Name < p[j].Name
+ }
+
+ // Newer versions first.
+ iSemVer, _ := semver.NewVersion(p[i].Version)
+ jSemVer, _ := semver.NewVersion(p[j].Version)
+ if iSemVer != nil && jSemVer != nil {
+ return jSemVer.LessThan(iSemVer)
+ }
+ return p[j].Version < p[i].Version
+}

indexer_test.go

@@ -0,0 +1,83 @@
+// 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 (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+
+ "github.com/elastic/package-registry/packages"
+)
+
+func TestLatestPackagesVersion(t *testing.T) {
+ newPackage := func(name string, version string) *packages.Package {
+ p := new(packages.Package)
+ p.Name = name
+ p.Version = version
+ return p
+ }
+
+ cases := []struct {
+ title string
+ source packages.Packages
+ expected packages.Packages
+ }{
+ {
+ title: "single package",
+ source: packages.Packages{
+ newPackage("foo", "1.2.3"),
+ },
+ expected: packages.Packages{
+ newPackage("foo", "1.2.3"),
+ },
+ },
+ {
+ title: "single package sorted versions",
+ source: packages.Packages{
+ newPackage("foo", "1.2.3"),
+ newPackage("foo", "1.2.2"),
+ newPackage("foo", "1.2.1"),
+ },
+ expected: packages.Packages{
+ newPackage("foo", "1.2.3"),
+ },
+ },
+ {
+ title: "single package unsorted versions",
+ source: packages.Packages{
+ newPackage("foo", "1.2.2"),
+ newPackage("foo", "1.2.1"),
+ newPackage("foo", "1.2.3"),
+ },
+ expected: packages.Packages{
+ newPackage("foo", "1.2.3"),
+ },
+ },
+ {
+ title: "multiple packages with multiple versions",
+ source: packages.Packages{
+ newPackage("foo", "1.2.2"),
+ newPackage("bar", "1.2.1"),
+ newPackage("bar", "1.2.2"),
+ newPackage("foo", "1.2.1"),
+ newPackage("bar", "1.2.3"),
+ newPackage("foo", "1.2.3"),
+ },
+ expected: packages.Packages{
+ newPackage("bar", "1.2.3"),
+ newPackage("foo", "1.2.3"),
+ },
+ },
+ }
+
+ for _, c := range cases {
+ t.Run(c.title, func(t *testing.T) {
+ result := latestPackagesVersion(c.source)
+ assert.EqualValues(t, c.expected, result)
+ })
+ }
+
+}

main.go

@@ -192,8 +192,8 @@ func initIndexers(ctx context.Context, logger *zap.Logger, config *Config) Combi
  WatchInterval: storageIndexerWatchInterval,
  }))
  } else {
- indexers = append(indexers, packages.NewFileSystemIndexer(packagesBasePaths...))
  indexers = append(indexers, packages.NewZipFileSystemIndexer(packagesBasePaths...))
+ indexers = append(indexers, packages.NewFileSystemIndexer(packagesBasePaths...))
  }
  combinedIndexer := NewCombinedIndexer(indexers...)
  ensurePackagesAvailable(ctx, logger, combinedIndexer)

main_test.go

@@ -42,7 +42,10 @@ func init() {
 
 func TestEndpoints(t *testing.T) {
  packagesBasePaths := []string{"./testdata/second_package_path", "./testdata/package"}
- indexer := packages.NewFileSystemIndexer(packagesBasePaths...)
+ indexer := NewCombinedIndexer(
+ packages.NewZipFileSystemIndexer("./testdata/local-storage"),
+ packages.NewFileSystemIndexer(packagesBasePaths...),
+ )
 
  err := indexer.Init(context.Background())
  require.NoError(t, err)
@@ -246,8 +249,8 @@ func TestStaticsModifiedTime(t *testing.T) {
  }
 
  indexer := NewCombinedIndexer(
- packages.NewFileSystemIndexer("./testdata/package"),
  packages.NewZipFileSystemIndexer("./testdata/local-storage"),
+ packages.NewFileSystemIndexer("./testdata/package"),
  )
  err := indexer.Init(context.Background())
  require.NoError(t, err)
@@ -307,8 +310,8 @@ func TestZippedArtifacts(t *testing.T) {
 
 func TestPackageIndex(t *testing.T) {
  indexer := NewCombinedIndexer(
- packages.NewFileSystemIndexer("./testdata/package"),
  packages.NewZipFileSystemIndexer("./testdata/local-storage"),
+ packages.NewFileSystemIndexer("./testdata/package"),
  )
 
  err := indexer.Init(context.Background())
@@ -425,8 +428,8 @@ func TestContentTypes(t *testing.T) {
  }
 
  indexer := NewCombinedIndexer(
- packages.NewFileSystemIndexer("./testdata/package"),
  packages.NewZipFileSystemIndexer("./testdata/local-storage"),
+ packages.NewFileSystemIndexer("./testdata/package"),
  )
 
  err := indexer.Init(context.Background())
@@ -454,8 +457,8 @@ func TestContentTypes(t *testing.T) {
 // on different file systems.
 func TestRangeDownloads(t *testing.T) {
  indexer := NewCombinedIndexer(
- packages.NewFileSystemIndexer("./testdata/package"),
  packages.NewZipFileSystemIndexer("./testdata/local-storage"),
+ packages.NewFileSystemIndexer("./testdata/package"),
  )
 
  err := indexer.Init(context.Background())

packages/packages.go

@@ -34,10 +34,43 @@ func (p Packages) Less(i, j int) bool {
  return p[i].Version < p[j].Version
 }
 
-// Join returns a set of packages that combines both sets.
+// Join returns a set of packages that combines both sets. If there is already
+// a package in `p1` with the same name and version that a package in `p2`, the
+// latter is not added.
 func (p1 Packages) Join(p2 Packages) Packages {
- // TODO: Avoid duplications?
- return append(p1, p2...)
+ for _, p := range p2 {
+ if p1.contains(p) {
+ continue
+ }
+ p1 = append(p1, p)
+ }
+ return p1
+}
+
+// contains returns true if `ps` contains a package with the same name and version as `p`.
+func (ps Packages) contains(p *Package) bool {
+ return ps.index(p) >= 0
+}
+
+// index finds if `ps` contains a package with the same name and version as `p` and
+// returns its index. If it is not found, it returns -1.
+func (ps Packages) index(p *Package) int {
+ for i, candidate := range ps {
+ if candidate.Name != p.Name {
+ continue
+ }
+ if cv, pv := candidate.versionSemVer, p.versionSemVer; cv != nil && pv != nil {
+ if !cv.Equal(pv) {
+ continue
+ }
+ }
+ if candidate.Version != p.Version {
+ continue
+ }
+
+ return i
+ }
+ return -1
 }
 
 // GetOptions can be used to pass options to Get.

packages/testdata/marshaler/packages.json

@@ -695,6 +695,102 @@
  }
  ]
  },
+ {
+ "name": "example",
+ "title": "Example Integration",
+ "version": "1.0.1",
+ "release": "ga",
+ "description": "This is the example integration",
+ "type": "integration",
+ "download": "/epr/example/example-1.0.1.zip",
+ "path": "/package/example/1.0.1",
+ "conditions": {
+ "kibana": {
+ "version": "~7.x.x"
+ }
+ },
+ "owner": {
+ "github": "ruflin"
+ },
+ "categories": [
+ "crm",
+ "azure"
+ ],
+ "format_version": "1.0.0",
+ "readme": "/package/example/1.0.1/docs/README.md",
+ "license": "basic",
+ "screenshots": [
+ {
+ "src": "/img/kibana-envoyproxy.jpg",
+ "path": "/package/example/1.0.1/img/kibana-envoyproxy.jpg",
+ "title": "IP Tables Ubiquity Dashboard",
+ "size": "1492x1464",
+ "type": "image/png"
+ }
+ ],
+ "assets": [
+ "/package/example/1.0.1/manifest.yml",
+ "/package/example/1.0.1/docs/README.md",
+ "/package/example/1.0.1/img/icon.png",
+ "/package/example/1.0.1/img/kibana-envoyproxy.jpg",
+ "/package/example/1.0.1/data_stream/foo/manifest.yml",
+ "/package/example/1.0.1/kibana/dashboard/0c610510-5cbd-11e9-8477-077ec9664dbd.json",
+ "/package/example/1.0.1/kibana/visualization/0a994af0-5c9d-11e9-8477-077ec9664dbd.json",
+ "/package/example/1.0.1/kibana/visualization/36f872a0-5c03-11e9-85b4-19d0072eb4f2.json",
+ "/package/example/1.0.1/kibana/visualization/38f96190-5c99-11e9-8477-077ec9664dbd.json",
+ "/package/example/1.0.1/kibana/visualization/7e4084e0-5c99-11e9-8477-077ec9664dbd.json",
+ "/package/example/1.0.1/kibana/visualization/80844540-5c97-11e9-8477-077ec9664dbd.json",
+ "/package/example/1.0.1/kibana/visualization/ab48c3f0-5ca6-11e9-8477-077ec9664dbd.json",
+ "/package/example/1.0.1/data_stream/foo/fields/base-fields.yml",
+ "/package/example/1.0.1/data_stream/foo/agent/stream/stream.yml.hbs",
+ "/package/example/1.0.1/data_stream/foo/elasticsearch/ingest_pipeline/pipeline-entry.json",
+ "/package/example/1.0.1/data_stream/foo/elasticsearch/ingest_pipeline/pipeline-http.json",
+ "/package/example/1.0.1/data_stream/foo/elasticsearch/ingest_pipeline/pipeline-json.json",
+ "/package/example/1.0.1/data_stream/foo/elasticsearch/ingest_pipeline/pipeline-plaintext.json",
+ "/package/example/1.0.1/data_stream/foo/elasticsearch/ingest_pipeline/pipeline-tcp.json"
+ ],
+ "policy_templates": [
+ {
+ "name": "logs",
+ "title": "Logs datasource",
+ "description": "Datasource for your log files.",
+ "inputs": [
+ {
+ "type": "foo"
+ }
+ ],
+ "multiple": true
+ }
+ ],
+ "data_streams": [
+ {
+ "type": "logs",
+ "dataset": "example.foo",
+ "title": "Foo",
+ "release": "ga",
+ "ingest_pipeline": "pipeline-entry",
+ "streams": [
+ {
+ "input": "foo",
+ "vars": [
+ {
+ "name": "paths",
+ "type": "text",
+ "description": "Path to log files to be collected",
+ "multi": true,
+ "required": true,
+ "show_user": false
+ }
+ ],
+ "template_path": "stream.yml.hbs",
+ "enabled": true
+ }
+ ],
+ "package": "example",
+ "path": "foo"
+ }
+ ]
+ },
  {
  "name": "example",
  "title": "Example Integration",

testdata/generated/categories-experimental.json

@@ -7,7 +7,7 @@
  {
  "id": "azure",
  "title": "Azure",
- "count": 1
+ "count": 2
  },
  {
  "id": "cloud",
@@ -22,7 +22,7 @@
  {
  "id": "crm",
  "title": "CRM",
- "count": 1
+ "count": 2
  },
  {
  "id": "custom",

testdata/generated/categories-include-policy-templates.json

@@ -2,12 +2,12 @@
  {
  "id": "azure",
  "title": "Azure",
- "count": 2
+ "count": 3
  },
  {
  "id": "crm",
  "title": "CRM",
- "count": 2
+ "count": 3
  },
  {
  "id": "custom",