Take into account stack subscription to install packages to install p…

…ackages via upload API

cmd/install.go

@@ -11,6 +11,7 @@ import (
 "github.com/spf13/cobra"
 
 "github.com/elastic/elastic-package/internal/cobraext"
+"github.com/elastic/elastic-package/internal/elasticsearch"
 "github.com/elastic/elastic-package/internal/install"
 "github.com/elastic/elastic-package/internal/kibana"
 "github.com/elastic/elastic-package/internal/packages"
@@ -60,16 +61,29 @@ func installCommandAction(cmd *cobra.Command, _ []string) error {
 }
 
 var opts []kibana.ClientOption
+var esOpts []elasticsearch.ClientOption
+
 tlsSkipVerify, _ := cmd.Flags().GetBool(cobraext.TLSSkipVerifyFlagName)
 if tlsSkipVerify {
 opts = append(opts, kibana.TLSSkipVerify())
+esOpts = append(esOpts, elasticsearch.OptionWithSkipTLSVerify())
 }
 
 kibanaClient, err := stack.NewKibanaClientFromProfile(profile, opts...)
 if err != nil {
 return fmt.Errorf("could not create kibana client: %w", err)
 }
 
+esClient, err := stack.NewElasticsearchClientFromProfile(profile, esOpts...)
+if err != nil {
+return fmt.Errorf("could not create kibana client: %w", err)
+}
+
+subscription, err := esClient.Subscription(cmd.Context())
+if err != nil {
+return fmt.Errorf("failed to get subscription from stack: %w", err)
+}
+
 if zipPathFile == "" && packageRootPath == "" {
 var found bool
 var err error
@@ -83,10 +97,11 @@ func installCommandAction(cmd *cobra.Command, _ []string) error {
 }
 
 installer, err := installer.NewForPackage(installer.Options{
-Kibana: kibanaClient,
-RootPath: packageRootPath,
-SkipValidation: skipValidation,
-ZipPath: zipPathFile,
+Kibana: kibanaClient,
+RootPath: packageRootPath,
+SkipValidation: skipValidation,
+ZipPath: zipPathFile,
+StackSubscription: subscription,
 })
 if err != nil {
 return fmt.Errorf("package installation failed: %w", err)

cmd/testrunner.go

@@ -166,6 +166,15 @@ func testRunnerAssetCommandAction(cmd *cobra.Command, args []string) error {
 return fmt.Errorf("can't create Kibana client: %w", err)
 }
 
+esClient, err := stack.NewElasticsearchClientFromProfile(profile)
+if err != nil {
+return fmt.Errorf("can't create Elasticsearch client: %w", err)
+}
+err = esClient.CheckHealth(ctx)
+if err != nil {
+return err
+}
+
 globalTestConfig, err := testrunner.ReadGlobalTestConfig(packageRootPath)
 if err != nil {
 return fmt.Errorf("failed to read global config: %w", err)
@@ -174,6 +183,7 @@ func testRunnerAssetCommandAction(cmd *cobra.Command, args []string) error {
 runner := asset.NewAssetTestRunner(asset.AssetTestRunnerOptions{
 PackageRootPath: packageRootPath,
 KibanaClient: kibanaClient,
+ESClient: esClient,
 GlobalTestConfig: globalTestConfig.Asset,
 WithCoverage: testCoverage,
 CoverageType: testCoverageFormat,
@@ -564,7 +574,7 @@ func testRunnerSystemCommandAction(cmd *cobra.Command, args []string) error {
 return fmt.Errorf("failed to read global config: %w", err)
 }
 
-runner := system.NewSystemTestRunner(system.SystemTestRunnerOptions{
+runner, err := system.NewSystemTestRunner(system.SystemTestRunnerOptions{
 Profile: profile,
 PackageRootPath: packageRootPath,
 KibanaClient: kibanaClient,
@@ -584,6 +594,9 @@ func testRunnerSystemCommandAction(cmd *cobra.Command, args []string) error {
 CoverageType: testCoverageFormat,
 CheckFailureStore: checkFailureStore,
 })
+if err != nil {
+return fmt.Errorf("failed to create system test runner: %w", err)
+}
 
 logger.Debugf("Running suite...")
 results, err := testrunner.RunSuite(ctx, runner)
@@ -677,6 +690,15 @@ func testRunnerPolicyCommandAction(cmd *cobra.Command, args []string) error {
 return fmt.Errorf("can't create Kibana client: %w", err)
 }
 
+esClient, err := stack.NewElasticsearchClientFromProfile(profile)
+if err != nil {
+return fmt.Errorf("can't create Elasticsearch client: %w", err)
+}
+err = esClient.CheckHealth(ctx)
+if err != nil {
+return err
+}
+
 manifest, err := packages.ReadPackageManifestFromPackageRoot(packageRootPath)
 if err != nil {
 return fmt.Errorf("reading package manifest failed (path: %s): %w", packageRootPath, err)
@@ -687,16 +709,20 @@ func testRunnerPolicyCommandAction(cmd *cobra.Command, args []string) error {
 return fmt.Errorf("failed to read global config: %w", err)
 }
 
-runner := policy.NewPolicyTestRunner(policy.PolicyTestRunnerOptions{
+runner, err := policy.NewPolicyTestRunner(policy.PolicyTestRunnerOptions{
 PackageRootPath: packageRootPath,
 KibanaClient: kibanaClient,
+ESClient: esClient,
 DataStreams: dataStreams,
 FailOnMissingTests: failOnMissing,
 GenerateTestResult: generateTestResult,
 GlobalTestConfig: globalTestConfig.Policy,
 WithCoverage: testCoverage,
 CoverageType: testCoverageFormat,
 })
+if err != nil {
+return fmt.Errorf("failed to create policy test runner: %w", err)
+}
 
 results, err := testrunner.RunSuite(ctx, runner)
 if err != nil {

internal/benchrunner/runners/common/elasticsearch.go

@@ -59,3 +59,30 @@ func CountDocsInDataStream(ctx context.Context, esapi *elasticsearch.API, dataSt
 
 return numHits, nil
 }
+
+func StackSubscription(ctx context.Context, esapi *elasticsearch.API) (string, error) {
+resp, err := esapi.License.Get(esapi.License.Get.WithContext(ctx))
+if err != nil {
+return "", fmt.Errorf("error getting subscription: %w", err)
+}
+defer resp.Body.Close()
+
+if resp.StatusCode != http.StatusOK {
+return "", fmt.Errorf("failed to get subscription: %s", resp.String())
+}
+
+type licenseResponse struct {
+License struct {
+Type string `json:"Type"`
+} `json:"license"`
+}
+
+var data licenseResponse
+err = json.NewDecoder(resp.Body).Decode(&data)
+if err != nil {
+return "", fmt.Errorf("error decoding subscription: %w", err)
+}
+
+return data.License.Type, nil
+
+}

internal/benchrunner/runners/rally/runner.go

@@ -484,11 +484,16 @@ func (r *runner) installPackageFromRegistry(ctx context.Context, packageName, pa
 }
 
 func (r *runner) installPackageFromPackageRoot(ctx context.Context) error {
+subscription, err := common.StackSubscription(ctx, r.options.ESAPI)
+if err != nil {
+return fmt.Errorf("failed to get stack subscription: %w", err)
+}
 logger.Debug("Installing package...")
 installer, err := installer.NewForPackage(installer.Options{
-Kibana: r.options.KibanaClient,
-RootPath: r.options.PackageRootPath,
-SkipValidation: true,
+Kibana: r.options.KibanaClient,
+RootPath: r.options.PackageRootPath,
+SkipValidation: true,
+StackSubscription: subscription,
 })
 if err != nil {
 return fmt.Errorf("failed to initialize package installer: %w", err)

internal/benchrunner/runners/stream/runner.go

@@ -251,11 +251,16 @@ func (r *runner) installPackage(ctx context.Context) error {
 }
 
 func (r *runner) installPackageFromPackageRoot(ctx context.Context) error {
+subscription, err := common.StackSubscription(ctx, r.options.ESAPI)
+if err != nil {
+return fmt.Errorf("failed to get stack subscription: %w", err)
+}
 logger.Debug("Installing package...")
 installer, err := installer.NewForPackage(installer.Options{
-Kibana: r.options.KibanaClient,
-RootPath: r.options.PackageRootPath,
-SkipValidation: true,
+Kibana: r.options.KibanaClient,
+RootPath: r.options.PackageRootPath,
+SkipValidation: true,
+StackSubscription: subscription,
 })
 
 if err != nil {

internal/elasticsearch/client.go

@@ -196,7 +196,7 @@ type Info struct {
 Version struct {
 Number string `json:"number"`
 BuildFlavor string `json:"build_flavor"`
-} `json:"version`
+} `json:"version"`
 }
 
 // Info gets cluster information and metadata.
@@ -220,6 +220,33 @@ func (client *Client) Info(ctx context.Context) (*Info, error) {
 return &info, nil
 }
 
+// Subscription gets cluster subscription.
+func (client *Client) Subscription(ctx context.Context) (string, error) {
+resp, err := client.Client.License.Get(client.Client.License.Get.WithContext(ctx))
+if err != nil {
+return "", fmt.Errorf("error getting subscription: %w", err)
+}
+defer resp.Body.Close()
+
+if resp.StatusCode != http.StatusOK {
+return "", fmt.Errorf("failed to get subscription: %s", resp.String())
+}
+
+type licenseResponse struct {
+License struct {
+Type string `json:"Type"`
+} `json:"license"`
+}
+
+var data licenseResponse
+err = json.NewDecoder(resp.Body).Decode(&data)
+if err != nil {
+return "", fmt.Errorf("error decoding subscription: %w", err)
+}
+
+return data.License.Type, nil
+}
+
 // IsFailureStoreAvailable checks if the failure store is available.
 func (client *Client) IsFailureStoreAvailable(ctx context.Context) (bool, error) {
 // FIXME: Using the low-level transport till the API SDK supports the failure store.

internal/packages/installer/factory.go

@@ -8,6 +8,7 @@ import (
 "context"
 "errors"
 "fmt"
+"path/filepath"
 
 "github.com/Masterminds/semver/v3"
 
@@ -18,7 +19,10 @@ import (
 "github.com/elastic/elastic-package/internal/validation"
 )
 
-var semver8_7_0 = semver.MustParse("8.7.0")
+var (
+semver8_7_0 = semver.MustParse("8.7.0")
+semver8_8_2 = semver.MustParse("8.8.2")
+)
 
 // Installer is responsible for installation/uninstallation of the package.
 type Installer interface {
@@ -30,7 +34,10 @@ type Installer interface {
 
 // Options are the parameters used to build an installer.
 type Options struct {
-Kibana *kibana.Client
+Kibana *kibana.Client
+
+StackSubscription string
+
 RootPath string
 ZipPath string
 SkipValidation bool
@@ -48,17 +55,33 @@ func NewForPackage(options Options) (Installer, error) {
 if options.RootPath == "" && options.ZipPath == "" {
 return nil, errors.New("missing package root path or pre-built zip package")
 }
+if options.StackSubscription == "" {
+return nil, errors.New("missing stack subscription")
+}
 
 version, err := kibanaVersion(options.Kibana)
 if err != nil {
 return nil, fmt.Errorf("failed to get kibana version: %w", err)
 }
 
-supportsZip := !version.LessThan(semver8_7_0)
 if options.ZipPath != "" {
-if !supportsZip {
-return nil, fmt.Errorf("not supported uploading zip packages in Kibana %s (%s required)", version, semver8_7_0)
+manifest, err := packages.ReadPackageManifestFromZipPackage(options.ZipPath)
+if err != nil {
+return nil, fmt.Errorf("failed to read package manifest: %w", err)
 }
+logger.Debugf("Root Path: %s", options.ZipPath)
+logger.Debugf("Subscription package: %s", manifest.Subscription())
+logger.Debugf("Subscription stack: %s", options.StackSubscription)
+supportsUploadZip := supportedUploadZip(options.StackSubscription, version)
+if !supportsUploadZip {
+if version.LessThan(semver8_7_0) {
+return nil, fmt.Errorf("not supported uploading zip packages in Kibana %s (%s required)", version, semver8_7_0)
+}
+if version.LessThan(semver8_8_2) {
+return nil, fmt.Errorf("not supported uploading zip packages in Kibana %s using subscription %s (%s required)", version, manifest.Subscription(), semver8_8_2)
+}
+}
+
 if !options.SkipValidation {
 logger.Debugf("Validating built .zip package (path: %s)", options.ZipPath)
 errs, skipped := validation.ValidateAndFilterFromZip(options.ZipPath)
@@ -73,22 +96,46 @@ func NewForPackage(options Options) (Installer, error) {
 return CreateForZip(options.Kibana, options.ZipPath)
 }
 
+pkgManifestPath := filepath.Join(options.RootPath, packages.PackageManifestFile)
+manifest, err := packages.ReadPackageManifest(pkgManifestPath)
+if err != nil {
+return nil, fmt.Errorf("failed to read package manifest: %w", err)
+}
+logger.Debugf("Root Path: %s", options.RootPath)
+logger.Debugf("Subscription package: %s", manifest.Subscription())
+logger.Debugf("Subscription stack: %s", options.StackSubscription)
+supportsUploadZip := supportedUploadZip(options.StackSubscription, version)
+
 target, err := builder.BuildPackage(builder.BuildOptions{
 PackageRoot: options.RootPath,
-CreateZip: supportsZip,
+CreateZip: supportsUploadZip,
 SignPackage: false,
 SkipValidation: options.SkipValidation,
 })
 if err != nil {
 return nil, fmt.Errorf("failed to build package: %v", err)
 }
 
-if supportsZip {
+if supportsUploadZip {
+logger.Debugf("supported zip")
 return CreateForZip(options.Kibana, target)
 }
+logger.Debugf("Not supported upload at all")
 return CreateForManifest(options.Kibana, target)
 }
 
+func supportedUploadZip(pkgSubscription string, kibanaVersion *semver.Version) bool {
+if kibanaVersion.LessThan(semver8_7_0) {
+return false
+}
+
+if kibanaVersion.LessThan(semver8_8_2) && pkgSubscription == "basic" {
+return false
+}
+
+return true
+}
+
 func kibanaVersion(kibana *kibana.Client) (*semver.Version, error) {
 version, err := kibana.Version()
 if err != nil {

internal/packages/packages.go

@@ -32,6 +32,8 @@ const (
 dataStreamTypeMetrics = "metrics"
 dataStreamTypeSynthetics = "synthetics"
 dataStreamTypeTraces = "traces"
+
+defaultSubscription = "basic"
 )
 
 // VarValue represents a variable value as defined in a package or data stream
@@ -153,6 +155,17 @@ type PackageManifest struct {
 Elasticsearch *Elasticsearch `config:"elasticsearch" json:"elasticsearch" yaml:"elasticsearch"`
 }
 
+func (p PackageManifest) Subscription() string {
+if p.Conditions.Elastic.Subscription != "" {
+return p.Conditions.Elastic.Subscription
+}
+if p.License != "" {
+return p.License
+}
+
+return defaultSubscription
+}
+
 type ManifestIndexTemplate struct {
 IngestPipeline *ManifestIngestPipeline `config:"ingest_pipeline" json:"ingest_pipeline" yaml:"ingest_pipeline"`
 Mappings *ManifestMappings `config:"mappings" json:"mappings" yaml:"mappings"`