From f89a5054cf0962004c09399e42c3e748a2175393 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Fri, 26 Jan 2024 10:41:25 +0100
Subject: [PATCH 01/15] Add logstash to serverless provider

---
 internal/kibana/fleet.go                      | 26 +++++++
 internal/serverless/project.go                | 22 +++++-
 .../_static/serverless-logstash.yml.tmpl      | 36 +++++++++
 internal/stack/resources.go                   |  3 +
 internal/stack/serverless.go                  | 75 ++++++++++++++++++-
 internal/stack/serverlessresources.go         | 21 ++++--
 6 files changed, 175 insertions(+), 8 deletions(-)
 create mode 100644 internal/stack/_static/serverless-logstash.yml.tmpl

diff --git a/internal/kibana/fleet.go b/internal/kibana/fleet.go
index 850fb7fc98..01552395df 100644
--- a/internal/kibana/fleet.go
+++ b/internal/kibana/fleet.go
@@ -11,6 +11,13 @@ import (
 	"net/http"
 )
 
+type FleetOutput struct {
+	ID    string   `json:"id,omitempty"`
+	Name  string   `json:"name,omitempty"`
+	Hosts []string `json:"hosts,omitempty"`
+	Type  string   `json:"type,omitempty"`
+}
+
 // DefaultFleetServerURL returns the default Fleet server configured in Kibana
 func (c *Client) DefaultFleetServerURL() (string, error) {
 	path := fmt.Sprintf("%s/fleet_server_hosts", FleetAPI)
@@ -43,3 +50,22 @@ func (c *Client) DefaultFleetServerURL() (string, error) {
 
 	return "", errors.New("could not find the fleet server URL for this project")
 }
+
+// AddFleetOutput adds an additional output to kibana eg., logstash
+func (c *Client) AddFleetOutput(fo FleetOutput) error {
+	reqBody, err := json.Marshal(fo)
+	if err != nil {
+		return fmt.Errorf("could not convert fleetOutput (request) to JSON: %w", err)
+	}
+
+	statusCode, respBody, err := c.post(fmt.Sprintf("%s/outputs", FleetAPI), reqBody)
+	if err != nil {
+		return fmt.Errorf("could not create fleet output: %w", err)
+	}
+
+	if statusCode != http.StatusOK {
+		return fmt.Errorf("could not get status data; API status code = %d; response body = %s", statusCode, respBody)
+	}
+
+	return nil
+}
diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index 2b911e3e37..b58d707c73 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -131,6 +131,21 @@ func (p *Project) DefaultFleetServerURL(kibanaClient *kibana.Client) (string, er
 	return fleetURL, nil
 }
 
+func (p *Project) AddLogstashFleetOutput(kibanaClient *kibana.Client) error {
+	logstashFleetOutput := kibana.FleetOutput{
+		Name:  "logstash-output",
+		ID:    "logstash-fleet-output",
+		Type:  "logstash",
+		Hosts: []string{"logstash:5044"},
+	}
+
+	if err := kibanaClient.AddFleetOutput(logstashFleetOutput); err != nil {
+		return fmt.Errorf("failed to add logstash fleet output: %w", err)
+	}
+
+	return nil
+}
+
 func (p *Project) getESHealth(ctx context.Context, elasticsearchClient *elasticsearch.Client) error {
 	return elasticsearchClient.CheckHealth(ctx)
 }
@@ -177,7 +192,7 @@ func (p *Project) getFleetHealth(ctx context.Context) error {
 	return nil
 }
 
-func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Client) error {
+func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Client, logstashEnabled bool) error {
 	systemPackages, err := registry.Production.Revisions("system", registry.SearchOptions{
 		KibanaVersion: strings.TrimSuffix(stackVersion, kibana.SNAPSHOT_SUFFIX),
 	})
@@ -196,6 +211,11 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 		Namespace:         "default",
 		MonitoringEnabled: []string{"logs", "metrics"},
 	}
+
+	if logstashEnabled {
+		policy.DataOutputID = "fleet-logstash-output"
+	}
+
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
 	if err != nil {
 		return fmt.Errorf("error while creating agent policy: %w", err)
diff --git a/internal/stack/_static/serverless-logstash.yml.tmpl b/internal/stack/_static/serverless-logstash.yml.tmpl
new file mode 100644
index 0000000000..0a2fd20f84
--- /dev/null
+++ b/internal/stack/_static/serverless-logstash.yml.tmpl
@@ -0,0 +1,36 @@
+version: '2.3'
+services:
+{{ $logstash_enabled := fact "logstash_enabled" }}
+{{ if eq $logstash_enabled "true" }}
+  logstash:
+    depends_on:
+      elasticsearch:
+        condition: service_healthy
+      kibana:
+        condition: service_healthy
+    image: ${LOGSTASH_IMAGE_REF}
+    healthcheck:
+      test: bin/logstash -t
+      interval: 60s
+      timeout: 50s
+      retries: 5
+    command: bash -c 'if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstashpipeline/  logstash.conf'
+    volumes:
+      - "../certs/logstash:/usr/share/logstash/config/certs"
+      - "../certs/elasticsearch/cert.pem:/usr/share/logstash/config/certs/elasticsearchpem"
+      - "./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro"
+    ports:
+       - "127.0.0.1:5044:5044"
+       - "127.0.0.1:9600:9600"
+    environment:
+      - xpack.monitoring.enabled=false
+      - ELASTIC_USER=elastic
+      - ELASTIC_PASSWORD=changeme
+      - ELASTIC_HOSTS=https://127.0.0.1:9200
+
+  logstash_is_ready:
+    image: tianon/true
+    depends_on:
+      logstash:
+        condition: service_healthy
+{{ end }}
diff --git a/internal/stack/resources.go b/internal/stack/resources.go
index c0a8b10a32..4c020299c5 100644
--- a/internal/stack/resources.go
+++ b/internal/stack/resources.go
@@ -35,6 +35,9 @@ const (
 	// LogstashConfigFile is the logstash config file.
 	LogstashConfigFile = "logstash.conf"
 
+	// LogstashEnvFile is the logstash docker compose file
+	LogstashComposeFile = "logstash.yml"
+
 	// KibanaHealthcheckFile is the kibana healthcheck.
 	KibanaHealthcheckFile = "kibana_healthcheck.sh"
 
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 02197a753d..17f972d7b7 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -105,6 +105,11 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 	}
 	project.Endpoints.Fleet = config.Parameters[paramServerlessFleetURL]
 
+	err = project.AddLogstashFleetOutput(sp.kibanaClient)
+	if err != nil {
+		return Config{}, fmt.Errorf("failed to add logstash fleet output: %w", err)
+	}
+
 	printUserConfig(options.Printer, config)
 
 	// update config with latest updates (e.g. fleet server url)
@@ -145,6 +150,11 @@ func (sp *serverlessProvider) currentProjectWithClientsAndFleetEndpoint(config C
 	}
 	project.Endpoints.Fleet = fleetURL
 
+	err = project.AddLogstashFleetOutput(sp.kibanaClient)
+	if err != nil {
+		return nil, fmt.Errorf("failed to add logstash fleet output: %w", err)
+	}
+
 	return project, nil
 }
 
@@ -242,6 +252,11 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("serverless project type not supported: %s", settings.Type)
 	}
 
+	logstashEnabled := false
+	if options.Profile.Config("stack.logstash_enabled", "false") == "true" {
+		logstashEnabled = true
+	}
+
 	var project *serverless.Project
 
 	project, err = sp.currentProject(config)
@@ -261,7 +276,8 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		}
 
 		logger.Infof("Creating agent policy")
-		err = project.CreateAgentPolicy(options.StackVersion, sp.kibanaClient)
+		err = project.CreateAgentPolicy(options.StackVersion, sp.kibanaClient, logstashEnabled)
+
 		if err != nil {
 			return fmt.Errorf("failed to create agent policy: %w", err)
 		}
@@ -279,6 +295,14 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("failed to start local agent: %w", err)
 	}
 
+	if logstashEnabled {
+		logger.Infof("Starting local logstash")
+		err = sp.startLocalLogstash(options, config)
+		if err != nil {
+			return fmt.Errorf("failed to start local logstash: %w", err)
+		}
+	}
+
 	return nil
 }
 
@@ -291,6 +315,11 @@ func (sp *serverlessProvider) localAgentComposeProject() (*compose.Project, erro
 	return compose.NewProject(sp.composeProjectName(), composeFile)
 }
 
+func (sp *serverlessProvider) localLogstashComposeProject() (*compose.Project, error) {
+	composeFile := sp.profile.Path(profileStackPath, LogstashComposeFile)
+	return compose.NewProject(sp.composeProjectName(), composeFile)
+}
+
 func (sp *serverlessProvider) startLocalAgent(options Options, config Config) error {
 	err := applyServerlessResources(sp.profile, options.StackVersion, config)
 	if err != nil {
@@ -325,6 +354,30 @@ func (sp *serverlessProvider) startLocalAgent(options Options, config Config) er
 	return nil
 }
 
+func (sp *serverlessProvider) startLocalLogstash(options Options, config Config) error {
+	err := applyServerlessResources(sp.profile, options.StackVersion, config)
+	if err != nil {
+		return fmt.Errorf("could not initialize compose files for local logstash: %w", err)
+	}
+
+	project, err := sp.localLogstashComposeProject()
+	if err != nil {
+		return fmt.Errorf("could not initialize local logstash compose project")
+	}
+
+	err = project.Build(compose.CommandOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to build images for local logstash: %w", err)
+	}
+
+	err = project.Up(compose.CommandOptions{ExtraArgs: []string{"-d"}})
+	if err != nil {
+		return fmt.Errorf("failed to start local logstash: %w", err)
+	}
+
+	return nil
+}
+
 func (sp *serverlessProvider) TearDown(options Options) error {
 	config, err := LoadConfig(sp.profile)
 	if err != nil {
@@ -339,6 +392,12 @@ func (sp *serverlessProvider) TearDown(options Options) error {
 		errs = fmt.Errorf("failed to destroy local agent: %w", err)
 	}
 
+	err = sp.destroyLocalLogstash()
+	if err != nil {
+		logger.Errorf("failed to destroy local logstash: %v", err)
+		errs = fmt.Errorf("failed to destroy local logstash: %w", err)
+	}
+
 	project, err := sp.currentProject(config)
 	if err != nil {
 		return fmt.Errorf("failed to find current project: %w", err)
@@ -371,6 +430,20 @@ func (sp *serverlessProvider) destroyLocalAgent() error {
 	return nil
 }
 
+func (sp *serverlessProvider) destroyLocalLogstash() error {
+	project, err := sp.localLogstashComposeProject()
+	if err != nil {
+		return fmt.Errorf("could not initialize local logstash compose project")
+	}
+
+	err = project.Down(compose.CommandOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to destroy local logstash: %w", err)
+	}
+
+	return nil
+}
+
 func (sp *serverlessProvider) Update(options Options) error {
 	return fmt.Errorf("not implemented")
 }
diff --git a/internal/stack/serverlessresources.go b/internal/stack/serverlessresources.go
index 76c1630d3d..5dcb47406d 100644
--- a/internal/stack/serverlessresources.go
+++ b/internal/stack/serverlessresources.go
@@ -26,6 +26,14 @@ var (
 			Path:    ElasticAgentEnvFile,
 			Content: staticSource.Template("_static/elastic-agent.env.tmpl"),
 		},
+		&resource.File{
+			Path:    LogstashConfigFile,
+			Content: staticSource.Template("_static/logstash.conf.tmpl"),
+		},
+		&resource.File{
+			Path:    LogstashComposeFile,
+			Content: staticSource.Template("_static/serverless-logstash.yml.tmpl"),
+		},
 	}
 )
 
@@ -39,12 +47,13 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 
 	resourceManager := resource.NewManager()
 	resourceManager.AddFacter(resource.StaticFacter{
-		"agent_version": stackVersion,
-		"agent_image":   appConfig.StackImageRefs(stackVersion).ElasticAgent,
-		"username":      config.ElasticsearchUsername,
-		"password":      config.ElasticsearchPassword,
-		"kibana_host":   config.KibanaHost,
-		"fleet_url":     config.Parameters[paramServerlessFleetURL],
+		"agent_version":    stackVersion,
+		"agent_image":      appConfig.StackImageRefs(stackVersion).ElasticAgent,
+		"username":         config.ElasticsearchUsername,
+		"password":         config.ElasticsearchPassword,
+		"kibana_host":      config.KibanaHost,
+		"fleet_url":        config.Parameters[paramServerlessFleetURL],
+		"logstash_enabled": profile.Config("stack.logstash_enabled", "false"),
 	})
 
 	os.MkdirAll(stackDir, 0755)

From b87d4af7b3500f91a031cda5ccfce2e7c11df317 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Mon, 29 Jan 2024 17:52:33 +0100
Subject: [PATCH 02/15] Fix pr comments

---
 ...mpl => serverless-docker-compose.yml.tmpl} | 25 +++++
 .../_static/serverless-elastic-agent.yml.tmpl | 26 -----
 internal/stack/resources.go                   |  3 -
 internal/stack/serverless.go                  | 94 ++++---------------
 internal/stack/serverlessresources.go         |  6 +-
 5 files changed, 42 insertions(+), 112 deletions(-)
 rename internal/stack/_static/{serverless-logstash.yml.tmpl => serverless-docker-compose.yml.tmpl} (65%)
 delete mode 100644 internal/stack/_static/serverless-elastic-agent.yml.tmpl

diff --git a/internal/stack/_static/serverless-logstash.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
similarity index 65%
rename from internal/stack/_static/serverless-logstash.yml.tmpl
rename to internal/stack/_static/serverless-docker-compose.yml.tmpl
index 0a2fd20f84..e689c34d46 100644
--- a/internal/stack/_static/serverless-logstash.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -1,5 +1,30 @@
 version: '2.3'
 services:
+  elastic-agent:
+    image: "{{ fact "agent_image" }}"
+    healthcheck:
+      test: "elastic-agent status"
+      timeout: 2s
+      start_period: 360s
+      retries: 180
+      interval: 5s
+    hostname: docker-fleet-agent
+    env_file: "./elastic-agent.env"
+    volumes:
+    - type: bind
+      source: ../../../tmp/service_logs/
+      target: /tmp/service_logs/
+    # Mount service_logs under /run too as a testing workaround for the journald input (see elastic-package#1235).
+    - type: bind
+      source: ../../../tmp/service_logs/
+      target: /run/service_logs/
+
+  elastic-agent_is_ready:
+    image: tianon/true
+    depends_on:
+      elastic-agent:
+        condition: service_healthy
+
 {{ $logstash_enabled := fact "logstash_enabled" }}
 {{ if eq $logstash_enabled "true" }}
   logstash:
diff --git a/internal/stack/_static/serverless-elastic-agent.yml.tmpl b/internal/stack/_static/serverless-elastic-agent.yml.tmpl
deleted file mode 100644
index 94744bb2b7..0000000000
--- a/internal/stack/_static/serverless-elastic-agent.yml.tmpl
+++ /dev/null
@@ -1,26 +0,0 @@
-version: '2.3'
-services:
-  elastic-agent:
-    image: "{{ fact "agent_image" }}"
-    healthcheck:
-      test: "elastic-agent status"
-      timeout: 2s
-      start_period: 360s
-      retries: 180
-      interval: 5s
-    hostname: docker-fleet-agent
-    env_file: "./elastic-agent.env"
-    volumes:
-    - type: bind
-      source: ../../../tmp/service_logs/
-      target: /tmp/service_logs/
-    # Mount service_logs under /run too as a testing workaround for the journald input (see elastic-package#1235).
-    - type: bind
-      source: ../../../tmp/service_logs/
-      target: /run/service_logs/
-
-  elastic-agent_is_ready:
-    image: tianon/true
-    depends_on:
-      elastic-agent:
-        condition: service_healthy
diff --git a/internal/stack/resources.go b/internal/stack/resources.go
index 4c020299c5..c0a8b10a32 100644
--- a/internal/stack/resources.go
+++ b/internal/stack/resources.go
@@ -35,9 +35,6 @@ const (
 	// LogstashConfigFile is the logstash config file.
 	LogstashConfigFile = "logstash.conf"
 
-	// LogstashEnvFile is the logstash docker compose file
-	LogstashComposeFile = "logstash.yml"
-
 	// KibanaHealthcheckFile is the kibana healthcheck.
 	KibanaHealthcheckFile = "kibana_healthcheck.sh"
 
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 17f972d7b7..75678a1443 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -150,11 +150,6 @@ func (sp *serverlessProvider) currentProjectWithClientsAndFleetEndpoint(config C
 	}
 	project.Endpoints.Fleet = fleetURL
 
-	err = project.AddLogstashFleetOutput(sp.kibanaClient)
-	if err != nil {
-		return nil, fmt.Errorf("failed to add logstash fleet output: %w", err)
-	}
-
 	return project, nil
 }
 
@@ -289,18 +284,10 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		printUserConfig(options.Printer, config)
 	}
 
-	logger.Infof("Starting local agent")
-	err = sp.startLocalAgent(options, config)
+	logger.Infof("Starting local services")
+	err = sp.startLocalServices(options, config)
 	if err != nil {
-		return fmt.Errorf("failed to start local agent: %w", err)
-	}
-
-	if logstashEnabled {
-		logger.Infof("Starting local logstash")
-		err = sp.startLocalLogstash(options, config)
-		if err != nil {
-			return fmt.Errorf("failed to start local logstash: %w", err)
-		}
+		return fmt.Errorf("failed to start local service: %w", err)
 	}
 
 	return nil
@@ -310,30 +297,25 @@ func (sp *serverlessProvider) composeProjectName() string {
 	return DockerComposeProjectName(sp.profile)
 }
 
-func (sp *serverlessProvider) localAgentComposeProject() (*compose.Project, error) {
+func (sp *serverlessProvider) localServiceComposeProject() (*compose.Project, error) {
 	composeFile := sp.profile.Path(profileStackPath, SnapshotFile)
 	return compose.NewProject(sp.composeProjectName(), composeFile)
 }
 
-func (sp *serverlessProvider) localLogstashComposeProject() (*compose.Project, error) {
-	composeFile := sp.profile.Path(profileStackPath, LogstashComposeFile)
-	return compose.NewProject(sp.composeProjectName(), composeFile)
-}
-
-func (sp *serverlessProvider) startLocalAgent(options Options, config Config) error {
+func (sp *serverlessProvider) startLocalServices(options Options, config Config) error {
 	err := applyServerlessResources(sp.profile, options.StackVersion, config)
 	if err != nil {
-		return fmt.Errorf("could not initialize compose files for local agent: %w", err)
+		return fmt.Errorf("could not initialize compose files for local service: %w", err)
 	}
 
-	project, err := sp.localAgentComposeProject()
+	project, err := sp.localServiceComposeProject()
 	if err != nil {
-		return fmt.Errorf("could not initialize local agent compose project")
+		return fmt.Errorf("could not initialize local service compose project")
 	}
 
 	err = project.Build(compose.CommandOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to build images for local agent: %w", err)
+		return fmt.Errorf("failed to build images for local service: %w", err)
 	}
 
 	err = project.Up(compose.CommandOptions{ExtraArgs: []string{"-d"}})
@@ -354,30 +336,6 @@ func (sp *serverlessProvider) startLocalAgent(options Options, config Config) er
 	return nil
 }
 
-func (sp *serverlessProvider) startLocalLogstash(options Options, config Config) error {
-	err := applyServerlessResources(sp.profile, options.StackVersion, config)
-	if err != nil {
-		return fmt.Errorf("could not initialize compose files for local logstash: %w", err)
-	}
-
-	project, err := sp.localLogstashComposeProject()
-	if err != nil {
-		return fmt.Errorf("could not initialize local logstash compose project")
-	}
-
-	err = project.Build(compose.CommandOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to build images for local logstash: %w", err)
-	}
-
-	err = project.Up(compose.CommandOptions{ExtraArgs: []string{"-d"}})
-	if err != nil {
-		return fmt.Errorf("failed to start local logstash: %w", err)
-	}
-
-	return nil
-}
-
 func (sp *serverlessProvider) TearDown(options Options) error {
 	config, err := LoadConfig(sp.profile)
 	if err != nil {
@@ -386,16 +344,10 @@ func (sp *serverlessProvider) TearDown(options Options) error {
 
 	var errs error
 
-	err = sp.destroyLocalAgent()
-	if err != nil {
-		logger.Errorf("failed to destroy local agent: %v", err)
-		errs = fmt.Errorf("failed to destroy local agent: %w", err)
-	}
-
-	err = sp.destroyLocalLogstash()
+	err = sp.destroyLocalService()
 	if err != nil {
-		logger.Errorf("failed to destroy local logstash: %v", err)
-		errs = fmt.Errorf("failed to destroy local logstash: %w", err)
+		logger.Errorf("failed to destroy local service: %v", err)
+		errs = fmt.Errorf("failed to destroy local service: %w", err)
 	}
 
 	project, err := sp.currentProject(config)
@@ -416,29 +368,15 @@ func (sp *serverlessProvider) TearDown(options Options) error {
 	return errs
 }
 
-func (sp *serverlessProvider) destroyLocalAgent() error {
-	project, err := sp.localAgentComposeProject()
-	if err != nil {
-		return fmt.Errorf("could not initialize local agent compose project")
-	}
-
-	err = project.Down(compose.CommandOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to destroy local agent: %w", err)
-	}
-
-	return nil
-}
-
-func (sp *serverlessProvider) destroyLocalLogstash() error {
-	project, err := sp.localLogstashComposeProject()
+func (sp *serverlessProvider) destroyLocalService() error {
+	project, err := sp.localServiceComposeProject()
 	if err != nil {
-		return fmt.Errorf("could not initialize local logstash compose project")
+		return fmt.Errorf("could not initialize local services compose project")
 	}
 
 	err = project.Down(compose.CommandOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to destroy local logstash: %w", err)
+		return fmt.Errorf("failed to destroy local services: %w", err)
 	}
 
 	return nil
diff --git a/internal/stack/serverlessresources.go b/internal/stack/serverlessresources.go
index 5dcb47406d..acc818123b 100644
--- a/internal/stack/serverlessresources.go
+++ b/internal/stack/serverlessresources.go
@@ -20,7 +20,7 @@ var (
 	serverlessStackResources = []resource.Resource{
 		&resource.File{
 			Path:    SnapshotFile,
-			Content: staticSource.Template("_static/serverless-elastic-agent.yml.tmpl"),
+			Content: staticSource.Template("_static/serverless-docker-compose.yml.tmpl"),
 		},
 		&resource.File{
 			Path:    ElasticAgentEnvFile,
@@ -30,10 +30,6 @@ var (
 			Path:    LogstashConfigFile,
 			Content: staticSource.Template("_static/logstash.conf.tmpl"),
 		},
-		&resource.File{
-			Path:    LogstashComposeFile,
-			Content: staticSource.Template("_static/serverless-logstash.yml.tmpl"),
-		},
 	}
 )
 

From b744397bf5b4516519c6760a665eafbdd5dce1b6 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Tue, 30 Jan 2024 10:08:40 +0100
Subject: [PATCH 03/15] Fix ES host and credentials for serverless

---
 internal/stack/_static/logstash.conf.tmpl     |  4 ++--
 .../serverless-docker-compose.yml.tmpl        |  8 +++----
 internal/stack/resources.go                   |  5 +++--
 internal/stack/serverless.go                  | 22 +++++++++----------
 internal/stack/serverlessresources.go         | 15 +++++++------
 5 files changed, 28 insertions(+), 26 deletions(-)

diff --git a/internal/stack/_static/logstash.conf.tmpl b/internal/stack/_static/logstash.conf.tmpl
index 98a3958edc..55f8e7e242 100644
--- a/internal/stack/_static/logstash.conf.tmpl
+++ b/internal/stack/_static/logstash.conf.tmpl
@@ -12,7 +12,7 @@ input {
 filter {
   elastic_integration {
     remove_field => ['@version']
-    hosts => ["https://elasticsearch:9200"]
+    hosts => [{{ fact "elasticsearch_host" }}]
     username => {{ fact "username" }}
     password => {{ fact "password" }}
     ssl_enabled => true
@@ -23,7 +23,7 @@ filter {
 
 output {
   elasticsearch {
-    hosts => ["https://elasticsearch:9200"]
+    hosts => [{{ fact "elasticsearch_host" }}]
     user => {{ fact "username" }}
     password => {{ fact "password" }}
     ssl_enabled => true
diff --git a/internal/stack/_static/serverless-docker-compose.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
index e689c34d46..7e3eeecbbd 100644
--- a/internal/stack/_static/serverless-docker-compose.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -42,16 +42,16 @@ services:
     command: bash -c 'if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstashpipeline/  logstash.conf'
     volumes:
       - "../certs/logstash:/usr/share/logstash/config/certs"
-      - "../certs/elasticsearch/cert.pem:/usr/share/logstash/config/certs/elasticsearchpem"
+      - "../certs/elasticsearch/cert.pem:/usr/share/logstash/config/certs/elasticsearch.pem"
       - "./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro"
     ports:
        - "127.0.0.1:5044:5044"
        - "127.0.0.1:9600:9600"
     environment:
       - xpack.monitoring.enabled=false
-      - ELASTIC_USER=elastic
-      - ELASTIC_PASSWORD=changeme
-      - ELASTIC_HOSTS=https://127.0.0.1:9200
+      - ELASTIC_USER={{ fact "username" }}
+      - ELASTIC_PASSWORD={{ fact "password" }}
+      - ELASTIC_HOSTS={{ fact "elasticsearch_host" }}
 
   logstash_is_ready:
     image: tianon/true
diff --git a/internal/stack/resources.go b/internal/stack/resources.go
index c0a8b10a32..0605fcbd17 100644
--- a/internal/stack/resources.go
+++ b/internal/stack/resources.go
@@ -123,8 +123,9 @@ func applyResources(profile *profile.Profile, stackVersion string) error {
 		"kibana_version":        stackVersion,
 		"agent_version":         stackVersion,
 
-		"kibana_host": "https://kibana:5601",
-		"fleet_url":   "https://fleet-server:8220",
+		"kibana_host":        "https://kibana:5601",
+		"fleet_url":          "https://fleet-server:8220",
+		"elasticsearch_host": "https://elasticsearch:9200",
 
 		"username": elasticsearchUsername,
 		"password": elasticsearchPassword,
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 75678a1443..0f05a1349b 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -287,7 +287,7 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 	logger.Infof("Starting local services")
 	err = sp.startLocalServices(options, config)
 	if err != nil {
-		return fmt.Errorf("failed to start local service: %w", err)
+		return fmt.Errorf("failed to start local services: %w", err)
 	}
 
 	return nil
@@ -297,7 +297,7 @@ func (sp *serverlessProvider) composeProjectName() string {
 	return DockerComposeProjectName(sp.profile)
 }
 
-func (sp *serverlessProvider) localServiceComposeProject() (*compose.Project, error) {
+func (sp *serverlessProvider) localServicesComposeProject() (*compose.Project, error) {
 	composeFile := sp.profile.Path(profileStackPath, SnapshotFile)
 	return compose.NewProject(sp.composeProjectName(), composeFile)
 }
@@ -305,17 +305,17 @@ func (sp *serverlessProvider) localServiceComposeProject() (*compose.Project, er
 func (sp *serverlessProvider) startLocalServices(options Options, config Config) error {
 	err := applyServerlessResources(sp.profile, options.StackVersion, config)
 	if err != nil {
-		return fmt.Errorf("could not initialize compose files for local service: %w", err)
+		return fmt.Errorf("could not initialize compose files for local services: %w", err)
 	}
 
-	project, err := sp.localServiceComposeProject()
+	project, err := sp.localServicesComposeProject()
 	if err != nil {
-		return fmt.Errorf("could not initialize local service compose project")
+		return fmt.Errorf("could not initialize local services compose project")
 	}
 
 	err = project.Build(compose.CommandOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to build images for local service: %w", err)
+		return fmt.Errorf("failed to build images for local services: %w", err)
 	}
 
 	err = project.Up(compose.CommandOptions{ExtraArgs: []string{"-d"}})
@@ -344,10 +344,10 @@ func (sp *serverlessProvider) TearDown(options Options) error {
 
 	var errs error
 
-	err = sp.destroyLocalService()
+	err = sp.destroyLocalServices()
 	if err != nil {
-		logger.Errorf("failed to destroy local service: %v", err)
-		errs = fmt.Errorf("failed to destroy local service: %w", err)
+		logger.Errorf("failed to destroy local services: %v", err)
+		errs = fmt.Errorf("failed to destroy local services: %w", err)
 	}
 
 	project, err := sp.currentProject(config)
@@ -368,8 +368,8 @@ func (sp *serverlessProvider) TearDown(options Options) error {
 	return errs
 }
 
-func (sp *serverlessProvider) destroyLocalService() error {
-	project, err := sp.localServiceComposeProject()
+func (sp *serverlessProvider) destroyLocalServices() error {
+	project, err := sp.localServicesComposeProject()
 	if err != nil {
 		return fmt.Errorf("could not initialize local services compose project")
 	}
diff --git a/internal/stack/serverlessresources.go b/internal/stack/serverlessresources.go
index acc818123b..3d16c61f92 100644
--- a/internal/stack/serverlessresources.go
+++ b/internal/stack/serverlessresources.go
@@ -43,13 +43,14 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 
 	resourceManager := resource.NewManager()
 	resourceManager.AddFacter(resource.StaticFacter{
-		"agent_version":    stackVersion,
-		"agent_image":      appConfig.StackImageRefs(stackVersion).ElasticAgent,
-		"username":         config.ElasticsearchUsername,
-		"password":         config.ElasticsearchPassword,
-		"kibana_host":      config.KibanaHost,
-		"fleet_url":        config.Parameters[paramServerlessFleetURL],
-		"logstash_enabled": profile.Config("stack.logstash_enabled", "false"),
+		"agent_version":      stackVersion,
+		"agent_image":        appConfig.StackImageRefs(stackVersion).ElasticAgent,
+		"elasticsearch_host": config.ElasticsearchHost,
+		"username":           config.ElasticsearchUsername,
+		"password":           config.ElasticsearchPassword,
+		"kibana_host":        config.KibanaHost,
+		"fleet_url":          config.Parameters[paramServerlessFleetURL],
+		"logstash_enabled":   profile.Config("stack.logstash_enabled", "false"),
 	})
 
 	os.MkdirAll(stackDir, 0755)

From b68735227a1a127152b951e85de45bf4e7d99165 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Tue, 30 Jan 2024 10:18:09 +0100
Subject: [PATCH 04/15] move config check to projectsettings

---
 internal/stack/serverless.go | 20 +++++++++-----------
 1 file changed, 9 insertions(+), 11 deletions(-)

diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 0f05a1349b..985458a52c 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -29,6 +29,7 @@ const (
 	configRegion          = "stack.serverless.region"
 	configProjectType     = "stack.serverless.type"
 	configElasticCloudURL = "stack.elastic_cloud.host"
+	configLogstashEnabled = "stack.logstash_enabled"
 
 	defaultRegion      = "aws-us-east-1"
 	defaultProjectType = "observability"
@@ -54,7 +55,8 @@ type projectSettings struct {
 	Region string
 	Type   string
 
-	StackVersion string
+	StackVersion    string
+	LogstashEnabled bool
 }
 
 func (sp *serverlessProvider) createProject(settings projectSettings, options Options, conf Config) (Config, error) {
@@ -203,10 +205,11 @@ func (sp *serverlessProvider) createClients(project *serverless.Project) error {
 
 func getProjectSettings(options Options) (projectSettings, error) {
 	s := projectSettings{
-		Name:         createProjectName(options),
-		Type:         options.Profile.Config(configProjectType, defaultProjectType),
-		Region:       options.Profile.Config(configRegion, defaultRegion),
-		StackVersion: options.StackVersion,
+		Name:            createProjectName(options),
+		Type:            options.Profile.Config(configProjectType, defaultProjectType),
+		Region:          options.Profile.Config(configRegion, defaultRegion),
+		StackVersion:    options.StackVersion,
+		LogstashEnabled: options.Profile.Config(configLogstashEnabled, "false") == "true",
 	}
 
 	return s, nil
@@ -247,11 +250,6 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("serverless project type not supported: %s", settings.Type)
 	}
 
-	logstashEnabled := false
-	if options.Profile.Config("stack.logstash_enabled", "false") == "true" {
-		logstashEnabled = true
-	}
-
 	var project *serverless.Project
 
 	project, err = sp.currentProject(config)
@@ -271,7 +269,7 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		}
 
 		logger.Infof("Creating agent policy")
-		err = project.CreateAgentPolicy(options.StackVersion, sp.kibanaClient, logstashEnabled)
+		err = project.CreateAgentPolicy(options.StackVersion, sp.kibanaClient, settings.LogstashEnabled)
 
 		if err != nil {
 			return fmt.Errorf("failed to create agent policy: %w", err)

From cad9fb7ea050386e27baab15a055e7c17130acd4 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Thu, 1 Feb 2024 15:23:25 +0100
Subject: [PATCH 05/15] Add connections between agent logstash and es

---
 internal/kibana/fleet.go                      |  2 +-
 internal/serverless/project.go                |  2 +-
 internal/stack/_static/logstash.conf.tmpl     |  4 +--
 .../serverless-docker-compose.yml.tmpl        | 11 ++-----
 .../_static/serverless-logstash.conf.tmpl     | 33 +++++++++++++++++++
 internal/stack/certs.go                       |  6 +++-
 internal/stack/certs_test.go                  |  4 +--
 internal/stack/resources.go                   |  2 +-
 internal/stack/serverless.go                  |  3 +-
 internal/stack/serverlessresources.go         | 32 ++++++++++++++++--
 10 files changed, 78 insertions(+), 21 deletions(-)
 create mode 100644 internal/stack/_static/serverless-logstash.conf.tmpl

diff --git a/internal/kibana/fleet.go b/internal/kibana/fleet.go
index 01552395df..e57b1871fb 100644
--- a/internal/kibana/fleet.go
+++ b/internal/kibana/fleet.go
@@ -64,7 +64,7 @@ func (c *Client) AddFleetOutput(fo FleetOutput) error {
 	}
 
 	if statusCode != http.StatusOK {
-		return fmt.Errorf("could not get status data; API status code = %d; response body = %s", statusCode, respBody)
+		return fmt.Errorf("could not add fleet output; API status code = %d; response body = %s", statusCode, respBody)
 	}
 
 	return nil
diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index b58d707c73..b73c4bc67e 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -213,7 +213,7 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 	}
 
 	if logstashEnabled {
-		policy.DataOutputID = "fleet-logstash-output"
+		policy.DataOutputID = "logstash-fleet-output"
 	}
 
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
diff --git a/internal/stack/_static/logstash.conf.tmpl b/internal/stack/_static/logstash.conf.tmpl
index 55f8e7e242..98a3958edc 100644
--- a/internal/stack/_static/logstash.conf.tmpl
+++ b/internal/stack/_static/logstash.conf.tmpl
@@ -12,7 +12,7 @@ input {
 filter {
   elastic_integration {
     remove_field => ['@version']
-    hosts => [{{ fact "elasticsearch_host" }}]
+    hosts => ["https://elasticsearch:9200"]
     username => {{ fact "username" }}
     password => {{ fact "password" }}
     ssl_enabled => true
@@ -23,7 +23,7 @@ filter {
 
 output {
   elasticsearch {
-    hosts => [{{ fact "elasticsearch_host" }}]
+    hosts => ["https://elasticsearch:9200"]
     user => {{ fact "username" }}
     password => {{ fact "password" }}
     ssl_enabled => true
diff --git a/internal/stack/_static/serverless-docker-compose.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
index 7e3eeecbbd..5243f40bac 100644
--- a/internal/stack/_static/serverless-docker-compose.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -18,6 +18,7 @@ services:
     - type: bind
       source: ../../../tmp/service_logs/
       target: /run/service_logs/
+    - "../certs/logstash/ca-cert.pem:/etc/ssl/certs/logstash.pem"
 
   elastic-agent_is_ready:
     image: tianon/true
@@ -28,21 +29,15 @@ services:
 {{ $logstash_enabled := fact "logstash_enabled" }}
 {{ if eq $logstash_enabled "true" }}
   logstash:
-    depends_on:
-      elasticsearch:
-        condition: service_healthy
-      kibana:
-        condition: service_healthy
-    image: ${LOGSTASH_IMAGE_REF}
+    image: "{{ fact "logstash_image" }}"
     healthcheck:
       test: bin/logstash -t
       interval: 60s
       timeout: 50s
       retries: 5
-    command: bash -c 'if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstashpipeline/  logstash.conf'
+    command: bash -c 'if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstash/pipeline/logstash.conf'
     volumes:
       - "../certs/logstash:/usr/share/logstash/config/certs"
-      - "../certs/elasticsearch/cert.pem:/usr/share/logstash/config/certs/elasticsearch.pem"
       - "./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro"
     ports:
        - "127.0.0.1:5044:5044"
diff --git a/internal/stack/_static/serverless-logstash.conf.tmpl b/internal/stack/_static/serverless-logstash.conf.tmpl
new file mode 100644
index 0000000000..a75543a159
--- /dev/null
+++ b/internal/stack/_static/serverless-logstash.conf.tmpl
@@ -0,0 +1,33 @@
+input {
+  elastic_agent {
+    port => 5044
+    ssl_enabled => true
+    ssl_certificate_authorities => ["/usr/share/logstash/config/certs/ca-cert.pem"]
+    ssl_certificate => "/usr/share/logstash/config/certs/cert.pem"
+    ssl_key => "/usr/share/logstash/config/certs/key.pem"
+  }
+}
+
+
+filter {
+  elastic_integration {
+    remove_field => ['@version']
+    hosts => ["{{ fact "elasticsearch_host" }}"]
+    username => {{ fact "username" }}
+    password => {{ fact "password" }}
+    ssl_enabled => true
+    ssl_verification_mode => "full"
+  }
+}
+
+
+output {
+  elasticsearch {
+    hosts => ["{{ fact "elasticsearch_host" }}"]
+    user => {{ fact "username" }}
+    password => {{ fact "password" }}
+    ssl_enabled => true
+    data_stream => "true"
+    document_id => "%{[@metadata][_ingest_document][id]}"
+  }
+}
diff --git a/internal/stack/certs.go b/internal/stack/certs.go
index 7416d6e072..a63b159536 100644
--- a/internal/stack/certs.go
+++ b/internal/stack/certs.go
@@ -26,6 +26,10 @@ var tlsServices = []string{
 	"logstash",
 }
 
+var tlsServicesServerless = []string{
+	"logstash",
+}
+
 var (
 	// CertificatesDirectory is the path to the certificates directory inside a profile.
 	CertificatesDirectory = "certs"
@@ -43,7 +47,7 @@ var (
 // initTLSCertificates initializes all the certificates needed to run the services
 // managed by elastic-package stack. It includes a CA, and a pair of keys and
 // certificates for each service.
-func initTLSCertificates(fileProvider string, profilePath string) ([]resource.Resource, error) {
+func initTLSCertificates(fileProvider string, profilePath string, tlsServices []string) ([]resource.Resource, error) {
 	certsDir := filepath.Join(profilePath, CertificatesDirectory)
 	caCertFile := filepath.Join(profilePath, string(CACertificateFile))
 	caKeyFile := filepath.Join(profilePath, string(CAKeyFile))
diff --git a/internal/stack/certs_test.go b/internal/stack/certs_test.go
index 2d7daf2eba..15dfd44c2c 100644
--- a/internal/stack/certs_test.go
+++ b/internal/stack/certs_test.go
@@ -22,7 +22,7 @@ func TestTLSCertsInitialization(t *testing.T) {
 	assert.Error(t, verifyTLSCertificates(caCertFile, caCertFile, caKeyFile, ""))
 
 	providerName := "test-file"
-	resources, err := initTLSCertificates(providerName, profilePath)
+	resources, err := initTLSCertificates(providerName, profilePath, tlsServices)
 	require.NoError(t, err)
 
 	resourceManager := resource.NewManager()
@@ -54,7 +54,7 @@ func TestTLSCertsInitialization(t *testing.T) {
 		assert.Error(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
 
 		// Check it is created again and is validated by the same CA.
-		resources, err := initTLSCertificates(providerName, profilePath)
+		resources, err := initTLSCertificates(providerName, profilePath, tlsServices)
 		require.NoError(t, err)
 		_, err = resourceManager.Apply(resources)
 		require.NoError(t, err)
diff --git a/internal/stack/resources.go b/internal/stack/resources.go
index 0605fcbd17..d34d0b660a 100644
--- a/internal/stack/resources.go
+++ b/internal/stack/resources.go
@@ -145,7 +145,7 @@ func applyResources(profile *profile.Profile, stackVersion string) error {
 	resourceManager.RegisterProvider("certs", &resource.FileProvider{
 		Prefix: profile.ProfilePath,
 	})
-	certResources, err := initTLSCertificates("certs", profile.ProfilePath)
+	certResources, err := initTLSCertificates("certs", profile.ProfilePath, tlsServices)
 	if err != nil {
 		return fmt.Errorf("failed to create TLS files: %w", err)
 	}
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 985458a52c..2e0cc3dd11 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -77,7 +77,6 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 		paramServerlessProjectID:   project.ID,
 		paramServerlessProjectType: project.Type,
 	}
-
 	config.ElasticsearchHost = project.Endpoints.Elasticsearch
 	config.KibanaHost = project.Endpoints.Kibana
 	config.ElasticsearchUsername = project.Credentials.Username
@@ -109,7 +108,7 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 
 	err = project.AddLogstashFleetOutput(sp.kibanaClient)
 	if err != nil {
-		return Config{}, fmt.Errorf("failed to add logstash fleet output: %w", err)
+		return Config{}, err
 	}
 
 	printUserConfig(options.Printer, config)
diff --git a/internal/stack/serverlessresources.go b/internal/stack/serverlessresources.go
index 3d16c61f92..f385c3fe66 100644
--- a/internal/stack/serverlessresources.go
+++ b/internal/stack/serverlessresources.go
@@ -28,7 +28,7 @@ var (
 		},
 		&resource.File{
 			Path:    LogstashConfigFile,
-			Content: staticSource.Template("_static/logstash.conf.tmpl"),
+			Content: staticSource.Template("_static/serverless-logstash.conf.tmpl"),
 		},
 	}
 )
@@ -45,7 +45,8 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 	resourceManager.AddFacter(resource.StaticFacter{
 		"agent_version":      stackVersion,
 		"agent_image":        appConfig.StackImageRefs(stackVersion).ElasticAgent,
-		"elasticsearch_host": config.ElasticsearchHost,
+		"logstash_image":     appConfig.StackImageRefs(stackVersion).Logstash,
+		"elasticsearch_host": esHostWithPort(config.ElasticsearchHost),
 		"username":           config.ElasticsearchUsername,
 		"password":           config.ElasticsearchPassword,
 		"kibana_host":        config.KibanaHost,
@@ -58,7 +59,19 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 		Prefix: stackDir,
 	})
 
-	results, err := resourceManager.Apply(serverlessStackResources)
+	resources := append([]resource.Resource{}, serverlessStackResources...)
+
+	// Keeping certificates in the profile directory for backwards compatibility reasons.
+	resourceManager.RegisterProvider("certs", &resource.FileProvider{
+		Prefix: profile.ProfilePath,
+	})
+	certResources, err := initTLSCertificates("certs", profile.ProfilePath, tlsServicesServerless)
+	if err != nil {
+		return fmt.Errorf("failed to create TLS files: %w", err)
+	}
+	resources = append(resources, certResources...)
+
+	results, err := resourceManager.Apply(resources)
 	if err != nil {
 		var errors []string
 		for _, result := range results {
@@ -71,3 +84,16 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 
 	return nil
 }
+
+// esHostWithPort checks if the es host has a port already added in the string , else adds 443
+// This is to mitigate a known issue in logstash - https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-serverless
+func esHostWithPort(host string) string {
+	// The ES host of the form https://esHost or https://esHost:port
+	splitHost := strings.Split(host, ":")
+
+	// https://esHost:port port already defined
+	if len(splitHost) > 2 {
+		return host
+	}
+	return host + ":443"
+}

From 283ab98829b27929c8f297369b168e8bad7f4f05 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Thu, 1 Feb 2024 15:25:38 +0100
Subject: [PATCH 06/15] Enable ssl in agent for local stack

---
 internal/stack/_static/docker-compose-stack.yml.tmpl | 1 +
 internal/stack/_static/logstash.conf.tmpl            | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/internal/stack/_static/docker-compose-stack.yml.tmpl b/internal/stack/_static/docker-compose-stack.yml.tmpl
index 6c15d56c1f..aafd37de02 100644
--- a/internal/stack/_static/docker-compose-stack.yml.tmpl
+++ b/internal/stack/_static/docker-compose-stack.yml.tmpl
@@ -142,6 +142,7 @@ services:
     env_file: "./elastic-agent.env"
     volumes:
     - "../certs/ca-cert.pem:/etc/ssl/certs/elastic-package.pem"
+    - "../certs/logstash/ca-cert.pem:/etc/ssl/certs/logstash.pem"
     - type: bind
       source: ../../../tmp/service_logs/
       target: /tmp/service_logs/
diff --git a/internal/stack/_static/logstash.conf.tmpl b/internal/stack/_static/logstash.conf.tmpl
index 98a3958edc..64d0fa5ea9 100644
--- a/internal/stack/_static/logstash.conf.tmpl
+++ b/internal/stack/_static/logstash.conf.tmpl
@@ -1,7 +1,7 @@
 input {
   elastic_agent {
     port => 5044
-    ssl_enabled => false
+    ssl_enabled => true
     ssl_certificate_authorities => ["/usr/share/logstash/config/certs/ca-cert.pem"]
     ssl_certificate => "/usr/share/logstash/config/certs/cert.pem"
     ssl_key => "/usr/share/logstash/config/certs/key.pem"

From 74abca7d435acffa184ac139627ed537768d6ffc Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Fri, 2 Feb 2024 15:33:44 +0100
Subject: [PATCH 07/15] Enable ssl communication between logstash and agent

---
 internal/certs/certs.go                       | 23 +++++++++--
 internal/kibana/fleet.go                      | 37 ++++++++++++++---
 internal/serverless/project.go                | 41 +++++++++++++++++--
 internal/stack/_static/logstash.conf.tmpl     |  2 +-
 .../serverless-docker-compose.yml.tmpl        |  2 +-
 .../_static/serverless-logstash.conf.tmpl     | 12 +++---
 internal/stack/certs.go                       | 16 ++++++--
 internal/stack/serverless.go                  |  8 +++-
 internal/stack/serverlessresources.go         | 12 +++---
 internal/stack/serverlessresources_test.go    | 31 ++++++++++++++
 10 files changed, 154 insertions(+), 30 deletions(-)
 create mode 100644 internal/stack/serverlessresources_test.go

diff --git a/internal/certs/certs.go b/internal/certs/certs.go
index b0b901e3e7..7d9e6839fc 100644
--- a/internal/certs/certs.go
+++ b/internal/certs/certs.go
@@ -100,7 +100,7 @@ func LoadCA(certFile, keyFile string) (*Issuer, error) {
 }
 
 func newCA(parent *Issuer) (*Issuer, error) {
-	cert, err := New(true, parent)
+	cert, err := New(true, false, parent)
 	if err != nil {
 		return nil, err
 	}
@@ -115,13 +115,19 @@ func (i *Issuer) IssueIntermediate() (*Issuer, error) {
 // Issue issues a certificate with the given options. This certificate
 // can be used to configure a TLS server.
 func (i *Issuer) Issue(opts ...Option) (*Certificate, error) {
-	return New(false, i, opts...)
+	return New(false, false, i, opts...)
+}
+
+// IssueClient issues a certificate with the given options. This certificate
+// can be used to configure a TLS client.
+func (i *Issuer) IssueClient(opts ...Option) (*Certificate, error) {
+	return New(false, true, i, opts...)
 }
 
 // NewSelfSignedCert issues a self-signed certificate with the given options.
 // This certificate can be used to configure a TLS server.
 func NewSelfSignedCert(opts ...Option) (*Certificate, error) {
-	return New(false, nil, opts...)
+	return New(false, false, nil, opts...)
 }
 
 // Option is a function that can modify a certificate template. To be used
@@ -140,7 +146,7 @@ func WithName(name string) Option {
 
 // New is the main helper to create a certificate, it is recommended to
 // use the more specific ones for specific use cases.
-func New(isCA bool, issuer *Issuer, opts ...Option) (*Certificate, error) {
+func New(isCA, isClient bool, issuer *Issuer, opts ...Option) (*Certificate, error) {
 	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate key: %w", err)
@@ -172,6 +178,15 @@ func New(isCA bool, issuer *Issuer, opts ...Option) (*Certificate, error) {
 		} else {
 			template.Subject.CommonName = "intermediate elastic-package CA"
 		}
+		// If the requester is a client we set clientAuth instead
+	} else if isClient {
+		template.ExtKeyUsage = []x509.ExtKeyUsage{
+			x509.ExtKeyUsageClientAuth,
+		}
+
+		// Include local hostname and ips as alternates in service certificates.
+		template.DNSNames = []string{"localhost"}
+		template.IPAddresses = []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}
 	} else {
 		template.ExtKeyUsage = []x509.ExtKeyUsage{
 			// Required for Chrome in OSX to show the "Proceed anyway" link.
diff --git a/internal/kibana/fleet.go b/internal/kibana/fleet.go
index e57b1871fb..116e8af128 100644
--- a/internal/kibana/fleet.go
+++ b/internal/kibana/fleet.go
@@ -12,10 +12,17 @@ import (
 )
 
 type FleetOutput struct {
-	ID    string   `json:"id,omitempty"`
-	Name  string   `json:"name,omitempty"`
-	Hosts []string `json:"hosts,omitempty"`
-	Type  string   `json:"type,omitempty"`
+	ID    string    `json:"id,omitempty"`
+	Name  string    `json:"name,omitempty"`
+	Hosts []string  `json:"hosts,omitempty"`
+	Type  string    `json:"type,omitempty"`
+	SSL   *AgentSSL `json:"ssl,omitempty"`
+}
+
+type AgentSSL struct {
+	Ca_authorities []string `json:"certificate_authorities,omitempty"`
+	Certificate    string   `json:"certificate,omitempty"`
+	Key            string   `json:"key,omitempty"`
 }
 
 // DefaultFleetServerURL returns the default Fleet server configured in Kibana
@@ -51,7 +58,27 @@ func (c *Client) DefaultFleetServerURL() (string, error) {
 	return "", errors.New("could not find the fleet server URL for this project")
 }
 
-// AddFleetOutput adds an additional output to kibana eg., logstash
+// UpdateFleetOutput updates an existing output to fleet
+// For example, to update ssl certificates etc.,
+func (c *Client) UpdateFleetOutput(fo FleetOutput, outputId string) error {
+	reqBody, err := json.Marshal(fo)
+	if err != nil {
+		return fmt.Errorf("could not convert fleetOutput (request) to JSON: %w", err)
+	}
+
+	statusCode, respBody, err := c.put(fmt.Sprintf("%s/outputs/%s", FleetAPI, outputId), reqBody)
+	if err != nil {
+		return fmt.Errorf("could not update fleet output: %w", err)
+	}
+
+	if statusCode != http.StatusOK {
+		return fmt.Errorf("could not update fleet output; API status code = %d; response body = %s", statusCode, respBody)
+	}
+
+	return nil
+}
+
+// AddFleetOutput adds an additional output to fleet eg., logstash
 func (c *Client) AddFleetOutput(fo FleetOutput) error {
 	reqBody, err := json.Marshal(fo)
 	if err != nil {
diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index b73c4bc67e..bb08c43667 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -11,12 +11,15 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"os"
+	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/elastic/elastic-package/internal/elasticsearch"
 	"github.com/elastic/elastic-package/internal/kibana"
 	"github.com/elastic/elastic-package/internal/logger"
+	"github.com/elastic/elastic-package/internal/profile"
 	"github.com/elastic/elastic-package/internal/registry"
 )
 
@@ -131,10 +134,10 @@ func (p *Project) DefaultFleetServerURL(kibanaClient *kibana.Client) (string, er
 	return fleetURL, nil
 }
 
-func (p *Project) AddLogstashFleetOutput(kibanaClient *kibana.Client) error {
+func (p *Project) AddLogstashFleetOutput(profile *profile.Profile, kibanaClient *kibana.Client) error {
 	logstashFleetOutput := kibana.FleetOutput{
 		Name:  "logstash-output",
-		ID:    "logstash-fleet-output",
+		ID:    "fleet-logstash-output",
 		Type:  "logstash",
 		Hosts: []string{"logstash:5044"},
 	}
@@ -146,6 +149,38 @@ func (p *Project) AddLogstashFleetOutput(kibanaClient *kibana.Client) error {
 	return nil
 }
 
+func (p *Project) UpdateLogstashFleetOutput(profile *profile.Profile, kibanaClient *kibana.Client) error {
+	certsDir := filepath.Join(profile.ProfilePath, "certs", "elastic-agent")
+
+	caFile, err := os.ReadFile(filepath.Join(certsDir, "ca-cert.pem"))
+	if err != nil {
+		return fmt.Errorf("failed to read ca certificate: %w", err)
+	}
+
+	certFile, err := os.ReadFile(filepath.Join(certsDir, "cert.pem"))
+	if err != nil {
+		return fmt.Errorf("failed to read client certificate: %w", err)
+	}
+
+	keyFile, err := os.ReadFile(filepath.Join(certsDir, "key.pem"))
+	if err != nil {
+		return fmt.Errorf("failed to read client certificate public key: %w", err)
+	}
+
+	logstashFleetOutput := kibana.FleetOutput{
+		SSL: &kibana.AgentSSL{
+			Ca_authorities: []string{string(caFile)},
+			Certificate:    string(certFile),
+			Key:            string(keyFile)},
+	}
+
+	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, "fleet-logstash-output"); err != nil {
+		return fmt.Errorf("failed to update logstash fleet output: %w", err)
+	}
+
+	return nil
+}
+
 func (p *Project) getESHealth(ctx context.Context, elasticsearchClient *elasticsearch.Client) error {
 	return elasticsearchClient.CheckHealth(ctx)
 }
@@ -213,7 +248,7 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 	}
 
 	if logstashEnabled {
-		policy.DataOutputID = "logstash-fleet-output"
+		policy.DataOutputID = "fleet-logstash-output"
 	}
 
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
diff --git a/internal/stack/_static/logstash.conf.tmpl b/internal/stack/_static/logstash.conf.tmpl
index 64d0fa5ea9..98a3958edc 100644
--- a/internal/stack/_static/logstash.conf.tmpl
+++ b/internal/stack/_static/logstash.conf.tmpl
@@ -1,7 +1,7 @@
 input {
   elastic_agent {
     port => 5044
-    ssl_enabled => true
+    ssl_enabled => false
     ssl_certificate_authorities => ["/usr/share/logstash/config/certs/ca-cert.pem"]
     ssl_certificate => "/usr/share/logstash/config/certs/cert.pem"
     ssl_key => "/usr/share/logstash/config/certs/key.pem"
diff --git a/internal/stack/_static/serverless-docker-compose.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
index 5243f40bac..63f9a541b5 100644
--- a/internal/stack/_static/serverless-docker-compose.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -35,7 +35,7 @@ services:
       interval: 60s
       timeout: 50s
       retries: 5
-    command: bash -c 'if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstash/pipeline/logstash.conf'
+    command: bash -c 'openssl pkcs8 -inform PEM -in /usr/share/logstash/config/certs/key.pem -topk8 -nocrypt -outform PEM -out /usr/share/logstash/config/certs/logstash.pkcs8.key && chmod 777 /usr/share/logstash/config/certs/logstash.pkcs8.key && if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstash/pipeline/logstash.conf'
     volumes:
       - "../certs/logstash:/usr/share/logstash/config/certs"
       - "./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro"
diff --git a/internal/stack/_static/serverless-logstash.conf.tmpl b/internal/stack/_static/serverless-logstash.conf.tmpl
index a75543a159..f9af56f91b 100644
--- a/internal/stack/_static/serverless-logstash.conf.tmpl
+++ b/internal/stack/_static/serverless-logstash.conf.tmpl
@@ -4,7 +4,8 @@ input {
     ssl_enabled => true
     ssl_certificate_authorities => ["/usr/share/logstash/config/certs/ca-cert.pem"]
     ssl_certificate => "/usr/share/logstash/config/certs/cert.pem"
-    ssl_key => "/usr/share/logstash/config/certs/key.pem"
+    ssl_key => "/usr/share/logstash/config/certs/logstash.pkcs8.key"
+    ssl_client_authentication => "required"
   }
 }
 
@@ -13,8 +14,8 @@ filter {
   elastic_integration {
     remove_field => ['@version']
     hosts => ["{{ fact "elasticsearch_host" }}"]
-    username => {{ fact "username" }}
-    password => {{ fact "password" }}
+    username => '{{ fact "username" }}'
+    password => '{{ fact "password" }}'
     ssl_enabled => true
     ssl_verification_mode => "full"
   }
@@ -24,10 +25,9 @@ filter {
 output {
   elasticsearch {
     hosts => ["{{ fact "elasticsearch_host" }}"]
-    user => {{ fact "username" }}
-    password => {{ fact "password" }}
+    user => '{{ fact "username" }}'
+    password => '{{ fact "password" }}'
     ssl_enabled => true
     data_stream => "true"
-    document_id => "%{[@metadata][_ingest_document][id]}"
   }
 }
diff --git a/internal/stack/certs.go b/internal/stack/certs.go
index a63b159536..548f4b9ac7 100644
--- a/internal/stack/certs.go
+++ b/internal/stack/certs.go
@@ -28,6 +28,7 @@ var tlsServices = []string{
 
 var tlsServicesServerless = []string{
 	"logstash",
+	"elastic-agent",
 }
 
 var (
@@ -144,9 +145,18 @@ func initServiceTLSCertificates(ca *certs.Issuer, caCertFile string, certFile, k
 		return certs.LoadCertificate(certFile, keyFile)
 	}
 
-	cert, err := ca.Issue(certs.WithName(service))
-	if err != nil {
-		return nil, fmt.Errorf("error initializing certificate for %q", service)
+	var cert *certs.Certificate
+	var err error
+	if service == "elastic-agent" {
+		cert, err = ca.IssueClient(certs.WithName(service))
+		if err != nil {
+			return nil, fmt.Errorf("error initializing certificate for %q", service)
+		}
+	} else {
+		cert, err = ca.Issue(certs.WithName(service))
+		if err != nil {
+			return nil, fmt.Errorf("error initializing certificate for %q", service)
+		}
 	}
 
 	return cert, nil
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 2e0cc3dd11..2d9ff3dac9 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -106,7 +106,7 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 	}
 	project.Endpoints.Fleet = config.Parameters[paramServerlessFleetURL]
 
-	err = project.AddLogstashFleetOutput(sp.kibanaClient)
+	err = project.AddLogstashFleetOutput(sp.profile, sp.kibanaClient)
 	if err != nil {
 		return Config{}, err
 	}
@@ -287,6 +287,12 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("failed to start local services: %w", err)
 	}
 
+	// Updating the output with ssl certificates created in startLocalServices
+	err = project.UpdateLogstashFleetOutput(sp.profile, sp.kibanaClient)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
diff --git a/internal/stack/serverlessresources.go b/internal/stack/serverlessresources.go
index f385c3fe66..4d6f307174 100644
--- a/internal/stack/serverlessresources.go
+++ b/internal/stack/serverlessresources.go
@@ -6,6 +6,7 @@ package stack
 
 import (
 	"fmt"
+	"net/url"
 	"os"
 	"path/filepath"
 	"strings"
@@ -88,12 +89,11 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 // esHostWithPort checks if the es host has a port already added in the string , else adds 443
 // This is to mitigate a known issue in logstash - https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-serverless
 func esHostWithPort(host string) string {
-	// The ES host of the form https://esHost or https://esHost:port
-	splitHost := strings.Split(host, ":")
+	url, _ := url.Parse(host)
 
-	// https://esHost:port port already defined
-	if len(splitHost) > 2 {
-		return host
+	if url.Port() == "" {
+		return host + ":443"
 	}
-	return host + ":443"
+
+	return host
 }
diff --git a/internal/stack/serverlessresources_test.go b/internal/stack/serverlessresources_test.go
new file mode 100644
index 0000000000..1837afeff1
--- /dev/null
+++ b/internal/stack/serverlessresources_test.go
@@ -0,0 +1,31 @@
+// 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 stack
+
+import (
+	"testing"
+)
+
+func TestEsHostWithPort(t *testing.T) {
+	tests := []struct {
+		name string
+		host string
+		want string
+	}{
+		{"host without port", "https://hostname", "https://hostname:443"},
+		{"host with port", "https://hostname:443", "https://hostname:443"},
+		{"host with differernt port", "https://hostname:9200", "https://hostname:9200"},
+		{"ipv6 host", "http://[2001:db8:1f70::999:de8:7648:6e8]:100/", "http://[2001:db8:1f70::999:de8:7648:6e8]:100/"},
+		{"ipv6 host without port", "http://[2001:db8:1f70::999:de8:7648:6e8]", "http://[2001:db8:1f70::999:de8:7648:6e8]:443"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := esHostWithPort(tt.host); got != tt.want {
+				t.Errorf("esHostWithPort() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

From ed6160e22a51dac4c0e436875de01a0b91a4fe80 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Mon, 5 Feb 2024 12:23:50 +0100
Subject: [PATCH 08/15] Create clients when a project already exists during
 boot up

---
 internal/stack/serverless.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 2d9ff3dac9..69ea97c66f 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -287,6 +287,13 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("failed to start local services: %w", err)
 	}
 
+	//Re-create clients if running on an existing project
+	if sp.kibanaClient == nil {
+		if err = sp.createClients(project); err != nil {
+			return err
+		}
+	}
+
 	// Updating the output with ssl certificates created in startLocalServices
 	err = project.UpdateLogstashFleetOutput(sp.profile, sp.kibanaClient)
 	if err != nil {

From f8680bc088b53cb0e012751bf1444dd9fe7e0ae4 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Mon, 5 Feb 2024 14:09:18 +0100
Subject: [PATCH 09/15] fix PR comments

---
 internal/stack/serverlessresources.go      | 9 +++++++--
 internal/stack/serverlessresources_test.go | 2 ++
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/internal/stack/serverlessresources.go b/internal/stack/serverlessresources.go
index 4d6f307174..8eb9b1896a 100644
--- a/internal/stack/serverlessresources.go
+++ b/internal/stack/serverlessresources.go
@@ -6,6 +6,7 @@ package stack
 
 import (
 	"fmt"
+	"net"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -89,10 +90,14 @@ func applyServerlessResources(profile *profile.Profile, stackVersion string, con
 // esHostWithPort checks if the es host has a port already added in the string , else adds 443
 // This is to mitigate a known issue in logstash - https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-serverless
 func esHostWithPort(host string) string {
-	url, _ := url.Parse(host)
+	url, err := url.Parse(host)
+	if err != nil {
+		return host
+	}
 
 	if url.Port() == "" {
-		return host + ":443"
+		url.Host = net.JoinHostPort(url.Hostname(), "443")
+		return url.String()
 	}
 
 	return host
diff --git a/internal/stack/serverlessresources_test.go b/internal/stack/serverlessresources_test.go
index 1837afeff1..707ea97eb2 100644
--- a/internal/stack/serverlessresources_test.go
+++ b/internal/stack/serverlessresources_test.go
@@ -19,6 +19,8 @@ func TestEsHostWithPort(t *testing.T) {
 		{"host with differernt port", "https://hostname:9200", "https://hostname:9200"},
 		{"ipv6 host", "http://[2001:db8:1f70::999:de8:7648:6e8]:100/", "http://[2001:db8:1f70::999:de8:7648:6e8]:100/"},
 		{"ipv6 host without port", "http://[2001:db8:1f70::999:de8:7648:6e8]", "http://[2001:db8:1f70::999:de8:7648:6e8]:443"},
+		{"host with path", "https://hostname/xyz", "https://hostname:443/xyz"},
+		{"ipv6 host with path", "https://[::1]/xyz", "https://[::1]:443/xyz"},
 	}
 
 	for _, tt := range tests {

From 9ef0ad3e230b4e9ac2b0debf0c3bd0c5faa002f6 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Mon, 5 Feb 2024 17:41:55 +0100
Subject: [PATCH 10/15] fix pr comments

---
 internal/serverless/project.go                | 10 +++++++---
 .../_static/docker-compose-stack.yml.tmpl     |  1 -
 .../serverless-docker-compose.yml.tmpl        |  2 +-
 internal/stack/serverless.go                  | 19 ++++++++++---------
 4 files changed, 18 insertions(+), 14 deletions(-)

diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index bb08c43667..406b5d65ba 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -23,6 +23,10 @@ import (
 	"github.com/elastic/elastic-package/internal/registry"
 )
 
+const (
+	FLEET_LOGSTASH_OUTPUT = "fleet-logstash-output"
+)
+
 // Project represents a serverless project
 type Project struct {
 	url    string
@@ -137,7 +141,7 @@ func (p *Project) DefaultFleetServerURL(kibanaClient *kibana.Client) (string, er
 func (p *Project) AddLogstashFleetOutput(profile *profile.Profile, kibanaClient *kibana.Client) error {
 	logstashFleetOutput := kibana.FleetOutput{
 		Name:  "logstash-output",
-		ID:    "fleet-logstash-output",
+		ID:    FLEET_LOGSTASH_OUTPUT,
 		Type:  "logstash",
 		Hosts: []string{"logstash:5044"},
 	}
@@ -174,7 +178,7 @@ func (p *Project) UpdateLogstashFleetOutput(profile *profile.Profile, kibanaClie
 			Key:            string(keyFile)},
 	}
 
-	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, "fleet-logstash-output"); err != nil {
+	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, FLEET_LOGSTASH_OUTPUT); err != nil {
 		return fmt.Errorf("failed to update logstash fleet output: %w", err)
 	}
 
@@ -248,7 +252,7 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 	}
 
 	if logstashEnabled {
-		policy.DataOutputID = "fleet-logstash-output"
+		policy.DataOutputID = FLEET_LOGSTASH_OUTPUT
 	}
 
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
diff --git a/internal/stack/_static/docker-compose-stack.yml.tmpl b/internal/stack/_static/docker-compose-stack.yml.tmpl
index aafd37de02..6c15d56c1f 100644
--- a/internal/stack/_static/docker-compose-stack.yml.tmpl
+++ b/internal/stack/_static/docker-compose-stack.yml.tmpl
@@ -142,7 +142,6 @@ services:
     env_file: "./elastic-agent.env"
     volumes:
     - "../certs/ca-cert.pem:/etc/ssl/certs/elastic-package.pem"
-    - "../certs/logstash/ca-cert.pem:/etc/ssl/certs/logstash.pem"
     - type: bind
       source: ../../../tmp/service_logs/
       target: /tmp/service_logs/
diff --git a/internal/stack/_static/serverless-docker-compose.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
index 63f9a541b5..508e71cc8e 100644
--- a/internal/stack/_static/serverless-docker-compose.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -18,7 +18,7 @@ services:
     - type: bind
       source: ../../../tmp/service_logs/
       target: /run/service_logs/
-    - "../certs/logstash/ca-cert.pem:/etc/ssl/certs/logstash.pem"
+    - "../certs/ca-cert.pem:/etc/ssl/certs/elastic-package.pem"
 
   elastic-agent_is_ready:
     image: tianon/true
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 69ea97c66f..c1cef1c763 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -287,19 +287,20 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("failed to start local services: %w", err)
 	}
 
-	//Re-create clients if running on an existing project
-	if sp.kibanaClient == nil {
-		if err = sp.createClients(project); err != nil {
+	if settings.LogstashEnabled {
+		//Re-create clients if running on an existing project
+		if sp.kibanaClient == nil {
+			if err = sp.createClients(project); err != nil {
+				return err
+			}
+		}
+		// Updating the output with ssl certificates created in startLocalServices
+		err = project.UpdateLogstashFleetOutput(sp.profile, sp.kibanaClient)
+		if err != nil {
 			return err
 		}
 	}
 
-	// Updating the output with ssl certificates created in startLocalServices
-	err = project.UpdateLogstashFleetOutput(sp.profile, sp.kibanaClient)
-	if err != nil {
-		return err
-	}
-
 	return nil
 }
 

From 5456fd678b7e060ea58e2bfcd5bd33e3b492f1c9 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Tue, 6 Feb 2024 16:51:12 +0100
Subject: [PATCH 11/15] Refactor and add code comment

---
 internal/stack/serverless.go | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index c1cef1c763..af12152c88 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -106,11 +106,6 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 	}
 	project.Endpoints.Fleet = config.Parameters[paramServerlessFleetURL]
 
-	err = project.AddLogstashFleetOutput(sp.profile, sp.kibanaClient)
-	if err != nil {
-		return Config{}, err
-	}
-
 	printUserConfig(options.Printer, config)
 
 	// update config with latest updates (e.g. fleet server url)
@@ -124,6 +119,11 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 		return Config{}, fmt.Errorf("not all services are healthy: %w", err)
 	}
 
+	err = project.AddLogstashFleetOutput(sp.profile, sp.kibanaClient)
+	if err != nil {
+		return Config{}, err
+	}
+
 	return config, nil
 }
 
@@ -295,6 +295,8 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 			}
 		}
 		// Updating the output with ssl certificates created in startLocalServices
+		// This is needed because the client ssl certificates are updated for every
+		// call to stack up even though the project is already setup.
 		err = project.UpdateLogstashFleetOutput(sp.profile, sp.kibanaClient)
 		if err != nil {
 			return err

From 2419b438176c68a61f6e23e13936d94f149ecc89 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Wed, 7 Feb 2024 11:45:40 +0100
Subject: [PATCH 12/15] fix pr comments

---
 internal/serverless/project.go                | 10 ++---
 .../serverless-docker-compose.yml.tmpl        |  1 +
 .../_static/serverless-logstash.conf.tmpl     |  1 -
 internal/stack/certs.go                       | 41 +++++++++++--------
 internal/stack/certs_test.go                  | 18 ++++----
 internal/stack/serverless.go                  |  8 ++--
 6 files changed, 43 insertions(+), 36 deletions(-)

diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index 406b5d65ba..ea067b2f77 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -24,7 +24,7 @@ import (
 )
 
 const (
-	FLEET_LOGSTASH_OUTPUT = "fleet-logstash-output"
+	FleetLogstshOutput = "fleet-logstash-output"
 )
 
 // Project represents a serverless project
@@ -141,7 +141,7 @@ func (p *Project) DefaultFleetServerURL(kibanaClient *kibana.Client) (string, er
 func (p *Project) AddLogstashFleetOutput(profile *profile.Profile, kibanaClient *kibana.Client) error {
 	logstashFleetOutput := kibana.FleetOutput{
 		Name:  "logstash-output",
-		ID:    FLEET_LOGSTASH_OUTPUT,
+		ID:    FleetLogstshOutput,
 		Type:  "logstash",
 		Hosts: []string{"logstash:5044"},
 	}
@@ -168,7 +168,7 @@ func (p *Project) UpdateLogstashFleetOutput(profile *profile.Profile, kibanaClie
 
 	keyFile, err := os.ReadFile(filepath.Join(certsDir, "key.pem"))
 	if err != nil {
-		return fmt.Errorf("failed to read client certificate public key: %w", err)
+		return fmt.Errorf("failed to read client certificate private key: %w", err)
 	}
 
 	logstashFleetOutput := kibana.FleetOutput{
@@ -178,7 +178,7 @@ func (p *Project) UpdateLogstashFleetOutput(profile *profile.Profile, kibanaClie
 			Key:            string(keyFile)},
 	}
 
-	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, FLEET_LOGSTASH_OUTPUT); err != nil {
+	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, FleetLogstshOutput); err != nil {
 		return fmt.Errorf("failed to update logstash fleet output: %w", err)
 	}
 
@@ -252,7 +252,7 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 	}
 
 	if logstashEnabled {
-		policy.DataOutputID = FLEET_LOGSTASH_OUTPUT
+		policy.DataOutputID = FleetLogstshOutput
 	}
 
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
diff --git a/internal/stack/_static/serverless-docker-compose.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
index 508e71cc8e..41cabd08ed 100644
--- a/internal/stack/_static/serverless-docker-compose.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -35,6 +35,7 @@ services:
       interval: 60s
       timeout: 50s
       retries: 5
+    # logstash expects the key in pkcs8 format. Hence converting the key.pem to pkcs8 format using openssl.
     command: bash -c 'openssl pkcs8 -inform PEM -in /usr/share/logstash/config/certs/key.pem -topk8 -nocrypt -outform PEM -out /usr/share/logstash/config/certs/logstash.pkcs8.key && chmod 777 /usr/share/logstash/config/certs/logstash.pkcs8.key && if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstash/pipeline/logstash.conf'
     volumes:
       - "../certs/logstash:/usr/share/logstash/config/certs"
diff --git a/internal/stack/_static/serverless-logstash.conf.tmpl b/internal/stack/_static/serverless-logstash.conf.tmpl
index f9af56f91b..d42f1d493e 100644
--- a/internal/stack/_static/serverless-logstash.conf.tmpl
+++ b/internal/stack/_static/serverless-logstash.conf.tmpl
@@ -5,7 +5,6 @@ input {
     ssl_certificate_authorities => ["/usr/share/logstash/config/certs/ca-cert.pem"]
     ssl_certificate => "/usr/share/logstash/config/certs/cert.pem"
     ssl_key => "/usr/share/logstash/config/certs/logstash.pkcs8.key"
-    ssl_client_authentication => "required"
   }
 }
 
diff --git a/internal/stack/certs.go b/internal/stack/certs.go
index 548f4b9ac7..86736185ca 100644
--- a/internal/stack/certs.go
+++ b/internal/stack/certs.go
@@ -16,19 +16,24 @@ import (
 	"github.com/elastic/elastic-package/internal/certs"
 )
 
+type tlsService struct {
+	Name     string
+	IsClient bool
+}
+
 // tlsServices is the list of server TLS certificates that will be
 // created in the given path.
-var tlsServices = []string{
-	"elasticsearch",
-	"kibana",
-	"package-registry",
-	"fleet-server",
-	"logstash",
+var tlsServices = []tlsService{
+	{Name: "elasticsearch"},
+	{Name: "kibana"},
+	{Name: "package-registry"},
+	{Name: "fleet-server"},
+	{Name: "logstash"},
 }
 
-var tlsServicesServerless = []string{
-	"logstash",
-	"elastic-agent",
+var tlsServicesServerless = []tlsService{
+	{Name: "logstash"},
+	{Name: "elastic-agent", IsClient: true},
 }
 
 var (
@@ -48,7 +53,7 @@ var (
 // initTLSCertificates initializes all the certificates needed to run the services
 // managed by elastic-package stack. It includes a CA, and a pair of keys and
 // certificates for each service.
-func initTLSCertificates(fileProvider string, profilePath string, tlsServices []string) ([]resource.Resource, error) {
+func initTLSCertificates(fileProvider string, profilePath string, tlsServices []tlsService) ([]resource.Resource, error) {
 	certsDir := filepath.Join(profilePath, CertificatesDirectory)
 	caCertFile := filepath.Join(profilePath, string(CACertificateFile))
 	caKeyFile := filepath.Join(profilePath, string(CAKeyFile))
@@ -74,7 +79,7 @@ func initTLSCertificates(fileProvider string, profilePath string, tlsServices []
 	}
 
 	for _, service := range tlsServices {
-		certsDir := filepath.Join(certsDir, service)
+		certsDir := filepath.Join(certsDir, service.Name)
 		caFile := filepath.Join(certsDir, "ca-cert.pem")
 		certFile := filepath.Join(certsDir, "cert.pem")
 		keyFile := filepath.Join(certsDir, "key.pem")
@@ -139,23 +144,23 @@ func initCA(certFile, keyFile string) (*certs.Issuer, error) {
 	return ca, nil
 }
 
-func initServiceTLSCertificates(ca *certs.Issuer, caCertFile string, certFile, keyFile, service string) (*certs.Certificate, error) {
-	if err := verifyTLSCertificates(caCertFile, certFile, keyFile, service); err == nil {
+func initServiceTLSCertificates(ca *certs.Issuer, caCertFile string, certFile, keyFile string, service tlsService) (*certs.Certificate, error) {
+	if err := verifyTLSCertificates(caCertFile, certFile, keyFile, service.Name); err == nil {
 		// Certificate already present and valid, load it.
 		return certs.LoadCertificate(certFile, keyFile)
 	}
 
 	var cert *certs.Certificate
 	var err error
-	if service == "elastic-agent" {
-		cert, err = ca.IssueClient(certs.WithName(service))
+	if service.IsClient {
+		cert, err = ca.IssueClient(certs.WithName(service.Name))
 		if err != nil {
-			return nil, fmt.Errorf("error initializing certificate for %q", service)
+			return nil, fmt.Errorf("error initializing certificate for %q", service.Name)
 		}
 	} else {
-		cert, err = ca.Issue(certs.WithName(service))
+		cert, err = ca.Issue(certs.WithName(service.Name))
 		if err != nil {
-			return nil, fmt.Errorf("error initializing certificate for %q", service)
+			return nil, fmt.Errorf("error initializing certificate for %q", service.Name)
 		}
 	}
 
diff --git a/internal/stack/certs_test.go b/internal/stack/certs_test.go
index 15dfd44c2c..b218d51ac3 100644
--- a/internal/stack/certs_test.go
+++ b/internal/stack/certs_test.go
@@ -35,23 +35,23 @@ func TestTLSCertsInitialization(t *testing.T) {
 	assert.NoError(t, verifyTLSCertificates(caCertFile, caCertFile, caKeyFile, ""))
 
 	for _, service := range tlsServices {
-		t.Run(service, func(t *testing.T) {
-			serviceCertFile := filepath.Join(profilePath, "certs", service, "cert.pem")
-			serviceKeyFile := filepath.Join(profilePath, "certs", service, "key.pem")
-			assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
+		t.Run(service.Name, func(t *testing.T) {
+			serviceCertFile := filepath.Join(profilePath, "certs", service.Name, "cert.pem")
+			serviceKeyFile := filepath.Join(profilePath, "certs", service.Name, "key.pem")
+			assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
 		})
 	}
 
 	t.Run("service certificate individually recreated", func(t *testing.T) {
 		service := tlsServices[0]
-		serviceCertFile := filepath.Join(profilePath, "certs", service, "cert.pem")
-		serviceKeyFile := filepath.Join(profilePath, "certs", service, "key.pem")
-		assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
+		serviceCertFile := filepath.Join(profilePath, "certs", service.Name, "cert.pem")
+		serviceKeyFile := filepath.Join(profilePath, "certs", service.Name, "key.pem")
+		assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
 
 		// Remove the certificate.
 		os.Remove(serviceCertFile)
 		os.Remove(serviceKeyFile)
-		assert.Error(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
+		assert.Error(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
 
 		// Check it is created again and is validated by the same CA.
 		resources, err := initTLSCertificates(providerName, profilePath, tlsServices)
@@ -59,6 +59,6 @@ func TestTLSCertsInitialization(t *testing.T) {
 		_, err = resourceManager.Apply(resources)
 		require.NoError(t, err)
 
-		assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
+		assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
 	})
 }
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index af12152c88..30029ff22b 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -119,9 +119,11 @@ func (sp *serverlessProvider) createProject(settings projectSettings, options Op
 		return Config{}, fmt.Errorf("not all services are healthy: %w", err)
 	}
 
-	err = project.AddLogstashFleetOutput(sp.profile, sp.kibanaClient)
-	if err != nil {
-		return Config{}, err
+	if settings.LogstashEnabled {
+		err = project.AddLogstashFleetOutput(sp.profile, sp.kibanaClient)
+		if err != nil {
+			return Config{}, err
+		}
 	}
 
 	return config, nil

From 7cfd60b680d427d23d2a4b5760043d8a8b48e180 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Wed, 7 Feb 2024 15:21:16 +0100
Subject: [PATCH 13/15] change variable and add comment

---
 internal/serverless/project.go                        | 11 ++++++-----
 .../stack/_static/serverless-docker-compose.yml.tmpl  |  2 ++
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index ea067b2f77..25a1a0d959 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -24,7 +24,7 @@ import (
 )
 
 const (
-	FleetLogstshOutput = "fleet-logstash-output"
+	FleetLogstashOutput = "fleet-logstash-output"
 )
 
 // Project represents a serverless project
@@ -141,7 +141,7 @@ func (p *Project) DefaultFleetServerURL(kibanaClient *kibana.Client) (string, er
 func (p *Project) AddLogstashFleetOutput(profile *profile.Profile, kibanaClient *kibana.Client) error {
 	logstashFleetOutput := kibana.FleetOutput{
 		Name:  "logstash-output",
-		ID:    FleetLogstshOutput,
+		ID:    FleetLogstashOutput,
 		Type:  "logstash",
 		Hosts: []string{"logstash:5044"},
 	}
@@ -175,10 +175,11 @@ func (p *Project) UpdateLogstashFleetOutput(profile *profile.Profile, kibanaClie
 		SSL: &kibana.AgentSSL{
 			Ca_authorities: []string{string(caFile)},
 			Certificate:    string(certFile),
-			Key:            string(keyFile)},
+			Key:            string(keyFile),
+		},
 	}
 
-	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, FleetLogstshOutput); err != nil {
+	if err := kibanaClient.UpdateFleetOutput(logstashFleetOutput, FleetLogstashOutput); err != nil {
 		return fmt.Errorf("failed to update logstash fleet output: %w", err)
 	}
 
@@ -252,7 +253,7 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 	}
 
 	if logstashEnabled {
-		policy.DataOutputID = FleetLogstshOutput
+		policy.DataOutputID = FleetLogstashOutput
 	}
 
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
diff --git a/internal/stack/_static/serverless-docker-compose.yml.tmpl b/internal/stack/_static/serverless-docker-compose.yml.tmpl
index 41cabd08ed..2ec85a1577 100644
--- a/internal/stack/_static/serverless-docker-compose.yml.tmpl
+++ b/internal/stack/_static/serverless-docker-compose.yml.tmpl
@@ -36,6 +36,8 @@ services:
       timeout: 50s
       retries: 5
     # logstash expects the key in pkcs8 format. Hence converting the key.pem to pkcs8 format using openssl.
+    # Also logstash-filter-elastic_integration plugin is installed by default to run ingest pipelines in logstash.
+    # elastic-package#1637 made improvements to enable logstash stats through port 9600.
     command: bash -c 'openssl pkcs8 -inform PEM -in /usr/share/logstash/config/certs/key.pem -topk8 -nocrypt -outform PEM -out /usr/share/logstash/config/certs/logstash.pkcs8.key && chmod 777 /usr/share/logstash/config/certs/logstash.pkcs8.key && if [[ ! $(bin/logstash-plugin list) ==  *"logstash-filter-elastic_integration"* ]]; then echo "Missing plugin  logstash-filter-elastic_integration, installing now" && bin/logstash-plugin install  logstash-filter-elastic_integration; fi && bin/logstash -f /usr/share/logstash/pipeline/logstash.conf'
     volumes:
       - "../certs/logstash:/usr/share/logstash/config/certs"

From 3b5256cf96882cd9002df54ea8452380f64189bf Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Thu, 8 Feb 2024 10:07:48 +0100
Subject: [PATCH 14/15] fix pr comments

---
 internal/certs/certs.go        | 4 ++--
 internal/serverless/project.go | 7 ++-----
 internal/stack/serverless.go   | 7 ++++++-
 3 files changed, 10 insertions(+), 8 deletions(-)

diff --git a/internal/certs/certs.go b/internal/certs/certs.go
index 7d9e6839fc..977a99af6f 100644
--- a/internal/certs/certs.go
+++ b/internal/certs/certs.go
@@ -328,8 +328,8 @@ func checkExpectedCertUsage(cert *x509.Certificate) error {
 	if !cert.IsCA {
 		// Required for Chrome in OSX to show the "Proceed anyway" link.
 		// https://stackoverflow.com/a/64309893/28855
-		if !containsExtKeyUsage(cert.ExtKeyUsage, x509.ExtKeyUsageServerAuth) {
-			return fmt.Errorf("missing server auth key usage in certificate")
+		if !(containsExtKeyUsage(cert.ExtKeyUsage, x509.ExtKeyUsageServerAuth) || containsExtKeyUsage(cert.ExtKeyUsage, x509.ExtKeyUsageClientAuth)) {
+			return fmt.Errorf("missing either of server/client auth key usage in certificate")
 		}
 	}
 
diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index 25a1a0d959..242c24d781 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -232,7 +232,7 @@ func (p *Project) getFleetHealth(ctx context.Context) error {
 	return nil
 }
 
-func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Client, logstashEnabled bool) error {
+func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Client, outputId string) error {
 	systemPackages, err := registry.Production.Revisions("system", registry.SearchOptions{
 		KibanaVersion: strings.TrimSuffix(stackVersion, kibana.SNAPSHOT_SUFFIX),
 	})
@@ -250,10 +250,7 @@ func (p *Project) CreateAgentPolicy(stackVersion string, kibanaClient *kibana.Cl
 		Description:       "Policy created by elastic-package",
 		Namespace:         "default",
 		MonitoringEnabled: []string{"logs", "metrics"},
-	}
-
-	if logstashEnabled {
-		policy.DataOutputID = FleetLogstashOutput
+		DataOutputID:      outputId,
 	}
 
 	newPolicy, err := kibanaClient.CreatePolicy(policy)
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 30029ff22b..2a64d98424 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -269,8 +269,13 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 			return fmt.Errorf("failed to retrieve latest project created: %w", err)
 		}
 
+		outputID := ""
+		if settings.LogstashEnabled {
+			outputID = serverless.FleetLogstashOutput
+		}
+
 		logger.Infof("Creating agent policy")
-		err = project.CreateAgentPolicy(options.StackVersion, sp.kibanaClient, settings.LogstashEnabled)
+		err = project.CreateAgentPolicy(options.StackVersion, sp.kibanaClient, outputID)
 
 		if err != nil {
 			return fmt.Errorf("failed to create agent policy: %w", err)

From 35161e2285fe8482780f34372f6dd4365521bec4 Mon Sep 17 00:00:00 2001
From: Bharat Pasupula <saibharatchandra.pasupula@elastic.co>
Date: Thu, 8 Feb 2024 13:51:12 +0100
Subject: [PATCH 15/15] fix pr comments

---
 internal/kibana/fleet.go       |  6 +--
 internal/serverless/project.go |  6 +--
 internal/stack/certs.go        | 17 +++++---
 internal/stack/certs_test.go   | 74 ++++++++++++++++++++--------------
 internal/stack/serverless.go   | 15 +++----
 5 files changed, 66 insertions(+), 52 deletions(-)

diff --git a/internal/kibana/fleet.go b/internal/kibana/fleet.go
index c9cbfe8ecb..869044bb12 100644
--- a/internal/kibana/fleet.go
+++ b/internal/kibana/fleet.go
@@ -21,9 +21,9 @@ type FleetOutput struct {
 }
 
 type AgentSSL struct {
-	Ca_authorities []string `json:"certificate_authorities,omitempty"`
-	Certificate    string   `json:"certificate,omitempty"`
-	Key            string   `json:"key,omitempty"`
+	CertificateAuthorities []string `json:"certificate_authorities,omitempty"`
+	Certificate            string   `json:"certificate,omitempty"`
+	Key                    string   `json:"key,omitempty"`
 }
 
 // DefaultFleetServerURL returns the default Fleet server configured in Kibana
diff --git a/internal/serverless/project.go b/internal/serverless/project.go
index 242c24d781..6d5e3940b8 100644
--- a/internal/serverless/project.go
+++ b/internal/serverless/project.go
@@ -173,9 +173,9 @@ func (p *Project) UpdateLogstashFleetOutput(profile *profile.Profile, kibanaClie
 
 	logstashFleetOutput := kibana.FleetOutput{
 		SSL: &kibana.AgentSSL{
-			Ca_authorities: []string{string(caFile)},
-			Certificate:    string(certFile),
-			Key:            string(keyFile),
+			CertificateAuthorities: []string{string(caFile)},
+			Certificate:            string(certFile),
+			Key:                    string(keyFile),
 		},
 	}
 
diff --git a/internal/stack/certs.go b/internal/stack/certs.go
index 86736185ca..8c2c264ae6 100644
--- a/internal/stack/certs.go
+++ b/internal/stack/certs.go
@@ -129,7 +129,7 @@ func certWriteToResource(resources []resource.Resource, fileProvider string, pro
 }
 
 func initCA(certFile, keyFile string) (*certs.Issuer, error) {
-	if err := verifyTLSCertificates(certFile, certFile, keyFile, ""); err == nil {
+	if err := verifyTLSCertificates(certFile, certFile, keyFile, tlsService{}); err == nil {
 		// Valid CA is already present, load it to check service certificates.
 		ca, err := certs.LoadCA(certFile, keyFile)
 		if err != nil {
@@ -145,7 +145,7 @@ func initCA(certFile, keyFile string) (*certs.Issuer, error) {
 }
 
 func initServiceTLSCertificates(ca *certs.Issuer, caCertFile string, certFile, keyFile string, service tlsService) (*certs.Certificate, error) {
-	if err := verifyTLSCertificates(caCertFile, certFile, keyFile, service.Name); err == nil {
+	if err := verifyTLSCertificates(caCertFile, certFile, keyFile, service); err == nil {
 		// Certificate already present and valid, load it.
 		return certs.LoadCertificate(certFile, keyFile)
 	}
@@ -167,7 +167,7 @@ func initServiceTLSCertificates(ca *certs.Issuer, caCertFile string, certFile, k
 	return cert, nil
 }
 
-func verifyTLSCertificates(caFile, certFile, keyFile, name string) error {
+func verifyTLSCertificates(caFile, certFile, keyFile string, service tlsService) error {
 	cert, err := certs.LoadCertificate(certFile, keyFile)
 	if err != nil {
 		return err
@@ -180,9 +180,16 @@ func verifyTLSCertificates(caFile, certFile, keyFile, name string) error {
 	options := x509.VerifyOptions{
 		Roots: certPool,
 	}
-	if name != "" {
-		options.DNSName = name
+	if service.Name != "" {
+		options.DNSName = service.Name
 	}
+
+	// By default ExtKeyUsageServerAuth is add to KeyUsages
+	// See https://github.com/golang/go/blob/master/src/crypto/x509/verify.go#L193-L195
+	if service.IsClient {
+		options.KeyUsages = []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}
+	}
+
 	err = cert.Verify(options)
 	if err != nil {
 		return err
diff --git a/internal/stack/certs_test.go b/internal/stack/certs_test.go
index b218d51ac3..07b6183a7c 100644
--- a/internal/stack/certs_test.go
+++ b/internal/stack/certs_test.go
@@ -15,50 +15,62 @@ import (
 )
 
 func TestTLSCertsInitialization(t *testing.T) {
+
+	tests := []struct {
+		name     string
+		services []tlsService
+	}{
+		{"tlsServices", tlsServices},
+		{"tlsServicesServerless", tlsServicesServerless},
+	}
 	profilePath := t.TempDir()
 	caCertFile := filepath.Join(profilePath, "certs", "ca-cert.pem")
 	caKeyFile := filepath.Join(profilePath, "certs", "ca-key.pem")
 
-	assert.Error(t, verifyTLSCertificates(caCertFile, caCertFile, caKeyFile, ""))
+	assert.Error(t, verifyTLSCertificates(caCertFile, caCertFile, caKeyFile, tlsService{}))
 
 	providerName := "test-file"
-	resources, err := initTLSCertificates(providerName, profilePath, tlsServices)
-	require.NoError(t, err)
-
 	resourceManager := resource.NewManager()
-	resourceManager.RegisterProvider(providerName, &resource.FileProvider{
-		Prefix: profilePath,
-	})
-	_, err = resourceManager.Apply(resources)
-	require.NoError(t, err)
 
-	assert.NoError(t, verifyTLSCertificates(caCertFile, caCertFile, caKeyFile, ""))
+	for _, tt := range tests {
+		resources, err := initTLSCertificates(providerName, profilePath, tt.services)
+		require.NoError(t, err)
 
-	for _, service := range tlsServices {
-		t.Run(service.Name, func(t *testing.T) {
-			serviceCertFile := filepath.Join(profilePath, "certs", service.Name, "cert.pem")
-			serviceKeyFile := filepath.Join(profilePath, "certs", service.Name, "key.pem")
-			assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
+		resourceManager.RegisterProvider(providerName, &resource.FileProvider{
+			Prefix: profilePath,
 		})
+		_, err = resourceManager.Apply(resources)
+		require.NoError(t, err)
+
+		assert.NoError(t, verifyTLSCertificates(caCertFile, caCertFile, caKeyFile, tlsService{}))
+		for _, service := range tt.services {
+			t.Run(service.Name, func(t *testing.T) {
+				serviceCertFile := filepath.Join(profilePath, "certs", service.Name, "cert.pem")
+				serviceKeyFile := filepath.Join(profilePath, "certs", service.Name, "key.pem")
+				assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
+			})
+		}
 	}
 
-	t.Run("service certificate individually recreated", func(t *testing.T) {
-		service := tlsServices[0]
-		serviceCertFile := filepath.Join(profilePath, "certs", service.Name, "cert.pem")
-		serviceKeyFile := filepath.Join(profilePath, "certs", service.Name, "key.pem")
-		assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
+	for _, tt := range tests {
+		t.Run("service certificate individually recreated", func(t *testing.T) {
+			service := tt.services[0]
+			serviceCertFile := filepath.Join(profilePath, "certs", service.Name, "cert.pem")
+			serviceKeyFile := filepath.Join(profilePath, "certs", service.Name, "key.pem")
+			assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
 
-		// Remove the certificate.
-		os.Remove(serviceCertFile)
-		os.Remove(serviceKeyFile)
-		assert.Error(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
+			// Remove the certificate.
+			os.Remove(serviceCertFile)
+			os.Remove(serviceKeyFile)
+			assert.Error(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
 
-		// Check it is created again and is validated by the same CA.
-		resources, err := initTLSCertificates(providerName, profilePath, tlsServices)
-		require.NoError(t, err)
-		_, err = resourceManager.Apply(resources)
-		require.NoError(t, err)
+			// Check it is created again and is validated by the same CA.
+			resources, err := initTLSCertificates(providerName, profilePath, tt.services)
+			require.NoError(t, err)
+			_, err = resourceManager.Apply(resources)
+			require.NoError(t, err)
 
-		assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service.Name))
-	})
+			assert.NoError(t, verifyTLSCertificates(caCertFile, serviceCertFile, serviceKeyFile, service))
+		})
+	}
 }
diff --git a/internal/stack/serverless.go b/internal/stack/serverless.go
index 2a64d98424..f58204c23d 100644
--- a/internal/stack/serverless.go
+++ b/internal/stack/serverless.go
@@ -253,6 +253,7 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 
 	var project *serverless.Project
 
+	isNewProject := false
 	project, err = sp.currentProject(config)
 	switch err {
 	default:
@@ -280,6 +281,7 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		if err != nil {
 			return fmt.Errorf("failed to create agent policy: %w", err)
 		}
+		isNewProject = true
 
 		// TODO: Ensuring a specific GeoIP database would make tests reproducible
 		// Currently geo ip files would be ignored when running pipeline tests
@@ -294,16 +296,9 @@ func (sp *serverlessProvider) BootUp(options Options) error {
 		return fmt.Errorf("failed to start local services: %w", err)
 	}
 
-	if settings.LogstashEnabled {
-		//Re-create clients if running on an existing project
-		if sp.kibanaClient == nil {
-			if err = sp.createClients(project); err != nil {
-				return err
-			}
-		}
-		// Updating the output with ssl certificates created in startLocalServices
-		// This is needed because the client ssl certificates are updated for every
-		// call to stack up even though the project is already setup.
+	// Updating the output with ssl certificates created in startLocalServices
+	// The certificates are updated only when a new project is created and logstash is enabled
+	if isNewProject && settings.LogstashEnabled {
 		err = project.UpdateLogstashFleetOutput(sp.profile, sp.kibanaClient)
 		if err != nil {
 			return err