Set ingress host correctly (#967)

* Rename events topic

* Apply executor-specific ingress config at the executor

* Remove unnecessary ingress config

cmd/eventsprinter/cmd/root.go

@@ -35,7 +35,7 @@ func RootCmd() *cobra.Command {
 	cmd.PersistentFlags().String("url", "pulsar://localhost:6650", "URL to connect to Pulsar on.")
 	cmd.PersistentFlags().Bool("verbose", false, "Print full event sequences.")
 	cmd.PersistentFlags().String("subscription", "eventsprinter", "Subscription to connect to Pulsar on.")
-	cmd.PersistentFlags().String("topic", "persistent://armada/armada/jobset-events", "Pulsar topic to subscribe to.")
+	cmd.PersistentFlags().String("topic", "persistent://armada/armada/events", "Pulsar topic to subscribe to.")
 
 	return cmd
 }

config/lookoutingester/config.yaml

@@ -14,7 +14,7 @@ postgres:
 pulsar:
   enabled: true
   URL: "pulsar://localhost:6650"
-  jobsetEventsTopic: "jobset-events"
+  jobsetEventsTopic: "persistent://armada/armada/events"
 
 paralellism: 1
 subscriptionName: "lookout-ingester"

e2e/pulsar_test/pulsar_test.go

@@ -33,7 +33,7 @@ import (
 
 // Pulsar configuration. Must be manually reconciled with changes to the test setup or Armada.
 const pulsarUrl = "pulsar://localhost:6650"
-const pulsarTopic = "persistent://armada/armada/jobset-events"
+const pulsarTopic = "persistent://armada/armada/events"
 const pulsarSubscription = "e2e-test"
 const armadaUrl = "localhost:50051"
 const armadaQueueName = "e2e-test-queue"

e2e/setup/pulsar/armada-config.yaml

@@ -1,7 +1,7 @@
 pulsar:
   enabled: true
   URL: "pulsar://pulsar:6650"
-  jobsetEventsTopic: "persistent://armada/armada/jobset-events"
+  jobsetEventsTopic: "persistent://armada/armada/events"
   redisFromPulsarSubscription: "RedisFromPulsar"
   pulsarFromPulsarSubscription: "PulsarFromPulsar"
   hostnameSuffix: "svc"

internal/armada/server.go

@@ -28,7 +28,6 @@ import (
 	"github.com/G-Research/armada/internal/common/health"
 	"github.com/G-Research/armada/internal/common/task"
 	"github.com/G-Research/armada/internal/common/util"
-	executorconfig "github.com/G-Research/armada/internal/executor/configuration"
 	"github.com/G-Research/armada/internal/lookout/postgres"
 	"github.com/G-Research/armada/internal/pgkeyvalue"
 	"github.com/G-Research/armada/internal/pulsarutils"
@@ -193,11 +192,6 @@ func Serve(config *configuration.ArmadaConfig, healthChecks *health.MultiChecker
 			QueueRepository: queueRepository,
 			Permissions:     permissions,
 			SubmitServer:    submitServer,
-			IngressConfig: &executorconfig.IngressConfiguration{
-				HostnameSuffix: config.Pulsar.HostnameSuffix,
-				CertNameSuffix: config.Pulsar.CertNameSuffix,
-				Annotations:    config.Pulsar.Annotations,
-			},
 		}
 		submitServerToRegister = pulsarSubmitServer
 

internal/armada/server/submit_to_log.go

@@ -20,7 +20,7 @@ import (
 	"github.com/G-Research/armada/internal/common/auth/permission"
 	"github.com/G-Research/armada/internal/common/eventutil"
 	"github.com/G-Research/armada/internal/common/requestid"
-	executorconfig "github.com/G-Research/armada/internal/executor/configuration"
+	"github.com/G-Research/armada/internal/executor/configuration"
 	"github.com/G-Research/armada/internal/pgkeyvalue"
 	"github.com/G-Research/armada/internal/pulsarutils/pulsarrequestid"
 	"github.com/G-Research/armada/pkg/api"
@@ -42,9 +42,8 @@ type PulsarSubmitServer struct {
 	QueueRepository repository.QueueRepository
 	// Fall back to the legacy submit server for queue administration endpoints.
 	SubmitServer *SubmitServer
-	// Used to create k8s objects from the submitted ingresses and services.
-	IngressConfig *executorconfig.IngressConfiguration
-	KVStore       *pgkeyvalue.PGKeyValueStore
+	// Used for job submission deduplication.
+	KVStore *pgkeyvalue.PGKeyValueStore
 }
 
 // TODO: Add input validation to make sure messages can be inserted to the database.
@@ -181,7 +180,11 @@ func (srv *PulsarSubmitServer) SubmitJobs(ctx context.Context, req *api.JobSubmi
 		// Users submit API-specific service and ingress objects.
 		// However, the log only accepts proper k8s objects.
 		// Hence, the API-specific objects must be converted to proper k8s objects.
-		err = eventutil.PopulateK8sServicesIngresses(apiJob, srv.IngressConfig)
+		//
+		// We use an empty ingress config here.
+		// The executor applies executor-specific information later.
+		// We only need this here because we're re-using code that was previously called by the executor.
+		err = eventutil.PopulateK8sServicesIngresses(apiJob, &configuration.IngressConfiguration{})
 		if err != nil {
 			return nil, err
 		}

internal/executor/job/submit.go

@@ -168,14 +168,33 @@ func (allocationService *SubmitService) populateServicesIngresses(job *api.Job,
 		k8sServices, k8sIngresses := util2.GenerateIngresses(job, pod, allocationService.podDefaults.Ingress)
 		job.K8SService = k8sServices
 		job.K8SIngress = k8sIngresses
+	} else {
+		// If K8SIngress and/or K8SService was already populated, it was populated by the Pulsar submit API.
+		// Because the submit API can't know which executor the services/ingresses will be created in,
+		// the executor has to provide all executor-specific information.
+		for _, ingress := range job.K8SIngress {
+			ingress.Annotations = util.MergeMaps(
+				ingress.Annotations,
+				allocationService.podDefaults.Ingress.Annotations,
+			)
+
+			// We need to use indexing here since Spec.Rules isn't pointers.
+			for i, _ := range ingress.Spec.Rules {
+				ingress.Spec.Rules[i].Host += allocationService.podDefaults.Ingress.HostnameSuffix
+			}
+
+			// We need to use indexing here since Spec.TLS isn't pointers.
+			for i, _ := range ingress.Spec.TLS {
+				ingress.Spec.TLS[i].SecretName += allocationService.podDefaults.Ingress.CertNameSuffix
+			}
+		}
 	}
 	if job.K8SService == nil {
 		job.K8SService = make([]*v1.Service, 0)
 	}
 	if job.K8SIngress == nil {
 		job.K8SIngress = make([]*networking.Ingress, 0)
 	}
-	return
 }
 
 func exposesPorts(job *api.Job, podSpec *v1.PodSpec) bool {

makefile

@@ -265,7 +265,7 @@ tests-e2e-setup: setup-cluster
 	docker run --rm -v ${PWD}:/go/src/armada -w /go/src/armada -e KUBECONFIG=/go/src/armada/.kube/config --network kind bitnami/kubectl:1.23 apply -f ./e2e/setup/namespace-with-anonymous-user.yaml
 
 	# Armada dependencies.
-	docker run -d --name pulsar -p 0.0.0.0:6650:6650 --network=kind apachepulsar/pulsar:2.9.1 bin/pulsar standalone
+	docker run -d --name pulsar -p 0.0.0.0:6650:6650 --network=kind apachepulsar/pulsar:2.9.2 bin/pulsar standalone
 	docker run -d --name nats --network=kind nats-streaming:0.24.5
 	docker run -d --name redis -p=6379:6379 --network=kind redis:6.2.6
 	docker run -d --name postgres --network=kind -p 5432:5432 -e POSTGRES_PASSWORD=psw postgres:14.2
@@ -274,7 +274,8 @@ tests-e2e-setup: setup-cluster
 	sleep 10 # pulsar-admin errors if the Pulsar server hasn't started up yet.
 	docker exec -it pulsar bin/pulsar-admin tenants create armada
 	docker exec -it pulsar bin/pulsar-admin namespaces create armada/armada
-	docker exec -it pulsar bin/pulsar-admin topics create-partitioned-topic persistent://armada/armada/jobset-events -p 100
+	docker exec -it pulsar bin/pulsar-admin topics delete-partitioned-topic persistent://armada/armada/events -f || true
+	docker exec -it pulsar bin/pulsar-admin topics create-partitioned-topic persistent://armada/armada/events -p 2
 
 	# Disable topic auto-creation to ensure an error is thrown on using the wrong topic
 	# (Pulsar automatically created the public tenant and default namespace).