feat: Return Kafka admin server URL in kas-fleet-manager Kafka Resource.

internal/kafka/internal/api/admin/private/api/openapi.yaml

@@ -43,6 +43,7 @@ paths:
           Each query can be ordered by any of the following `kafkaRequests` fields:
 
           * bootstrap_server_host
+          * admin_api_server_url
           * cloud_provider
           * cluster_id
           * created_at

internal/kafka/internal/api/admin/private/api_default.go

@@ -274,7 +274,7 @@ GetKafkas Returns a list of Kafkas
  * @param optional nil or *GetKafkasOpts - Optional Parameters:
  * @param "Page" (optional.String) -  Page index
  * @param "Size" (optional.String) -  Number of items in each page
- * @param "OrderBy" (optional.String) -  Specifies the order by criteria. The syntax of this parameter is similar to the syntax of the `order by` clause of an SQL statement. Each query can be ordered by any of the following `kafkaRequests` fields:  * bootstrap_server_host * cloud_provider * cluster_id * created_at * href * id * instance_type * multi_az * name * organisation_id * owner * reauthentication_enabled * region * status * updated_at * version  For example, to return all Kafka instances ordered by their name, use the following syntax:  ```sql name asc ```  To return all Kafka instances ordered by their name _and_ created date, use the following syntax:  ```sql name asc, created_at asc ```  If the parameter isn't provided, or if the value is empty, then the results are ordered by name.
+ * @param "OrderBy" (optional.String) -  Specifies the order by criteria. The syntax of this parameter is similar to the syntax of the `order by` clause of an SQL statement. Each query can be ordered by any of the following `kafkaRequests` fields:  * bootstrap_server_host * admin_api_server_url * cloud_provider * cluster_id * created_at * href * id * instance_type * multi_az * name * organisation_id * owner * reauthentication_enabled * region * status * updated_at * version  For example, to return all Kafka instances ordered by their name, use the following syntax:  ```sql name asc ```  To return all Kafka instances ordered by their name _and_ created date, use the following syntax:  ```sql name asc, created_at asc ```  If the parameter isn't provided, or if the value is empty, then the results are ordered by name.
  * @param "Search" (optional.String) -  Search criteria.  The syntax of this parameter is similar to the syntax of the `where` clause of an SQL statement. Allowed fields in the search are `cloud_provider`, `name`, `owner`, `region`, and `status`. Allowed comparators are `<>`, `=`, or `LIKE`. Allowed joins are `AND` and `OR`. However, you can use a maximum of 10 joins in a search query.  Examples:  To return a Kafka instance with the name `my-kafka` and the region `aws`, use the following syntax:  ``` name = my-kafka and cloud_provider = aws ```[p-]  To return a Kafka instance with a name that starts with `my`, use the following syntax:  ``` name like my%25 ```  If the parameter isn't provided, or if the value is empty, then all the Kafka instances that the user has permission to see are returned.  Note. If the query is invalid, an error is returned.
 @return KafkaList
 */

internal/kafka/internal/api/dbapi/data_plane_kafka_status.go

@@ -12,6 +12,7 @@ type DataPlaneKafkaStatus struct {
 	KafkaVersion    string
 	StrimziVersion  string
 	KafkaIBPVersion string
+	AdminServerURI  string
 }
 
 type DataPlaneKafkaStatusCondition struct {

internal/kafka/internal/api/dbapi/kafka_request_types.go

@@ -21,6 +21,7 @@ type KafkaRequest struct {
 	Owner                            string `json:"owner" gorm:"index"` // TODO: ocm owner?
 	OwnerAccountId                   string `json:"owner_account_id"`
 	BootstrapServerHost              string `json:"bootstrap_server_host"`
+	AdminApiServerURL                string `json:"admin_api_server_url"`
 	OrganisationId                   string `json:"organisation_id" gorm:"index"`
 	FailedReason                     string `json:"failed_reason"`
 	// PlacementId field should be updated every time when a KafkaRequest is assigned to an OSD cluster (even if it's the same one again)

internal/kafka/internal/api/private/api/openapi.yaml

@@ -441,6 +441,8 @@ components:
             $ref: '#/components/schemas/DataPlaneKafkaStatus_routes'
           nullable: true
           type: array
+        adminServerURI:
+          type: string
       type: object
     DataPlaneKafkaStatusUpdateRequest:
       additionalProperties:

internal/kafka/internal/api/private/model_data_plane_kafka_status.go

@@ -16,5 +16,6 @@ type DataPlaneKafkaStatus struct {
 	Capacity   DataPlaneKafkaStatusCapacity                    `json:"capacity,omitempty"`
 	Versions   DataPlaneKafkaStatusVersions                    `json:"versions,omitempty"`
 	// Routes created for a Kafka cluster
-	Routes *[]DataPlaneKafkaStatusRoutes `json:"routes,omitempty"`
+	Routes         *[]DataPlaneKafkaStatusRoutes `json:"routes,omitempty"`
+	AdminServerURI string                        `json:"adminServerURI,omitempty"`
 }

internal/kafka/internal/api/public/api/openapi.yaml

@@ -304,6 +304,7 @@ paths:
           Each query can be ordered by any of the following `kafkaRequests` fields:
 
           * bootstrap_server_host
+          * admin_api_server_url
           * cloud_provider
           * cluster_id
           * created_at
@@ -1206,6 +1207,7 @@ components:
         owner: api_kafka_service
         name: serviceapi
         bootstrap_server_host: serviceapi-1isy6rq3jki8q0otmjqfd3ocfrg.apps.mk-bttg0jn170hp.x5u8.s1.devshift.org
+        admin_api_server_url: https://admin-server-mk-e-e-e-e-c---{}ld{-}-n-vp--bltg.rhcloud.com
         created_at: 2020-10-05T12:51:24.053142Z
         updated_at: 2020-10-05T12:56:36.362208Z
         version: 2.6.0
@@ -1231,6 +1233,7 @@ components:
         owner: api_kafka_service
         name: serviceapi
         bootstrap_server_host: serviceapi-1isy6rq3jki8q0otmjqfd3ocfrg.apps.mk-bttg0jn170hp.x5u8.s1.devshift.org
+        admin_api_server_url: https://admin-server-mk-e-e-e-e-c---{}ld{-}-n-vp--bltg.rhcloud.com
         created_at: 2020-10-05T12:51:24.053142Z
         updated_at: 2020-10-05T12:56:36.362208Z
         failed_reason: a reason the Kafka request creation failed
@@ -1563,6 +1566,7 @@ components:
         Each query can be ordered by any of the following `kafkaRequests` fields:
 
         * bootstrap_server_host
+        * admin_api_server_url
         * cloud_provider
         * cluster_id
         * created_at
@@ -2033,6 +2037,11 @@ components:
           type: string
         bootstrap_server_host:
           type: string
+        admin_api_server_url:
+          description: The kafka admin server url to perform kafka admin operations
+            e.g acl management etc. The value will be available when the Kafka has
+            been fully provisioned i.e it reaches a 'ready' state
+          type: string
         created_at:
           format: date-time
           type: string

internal/kafka/internal/api/public/api_default.go

@@ -872,7 +872,7 @@ GetKafkas Returns a list of Kafka requests
  * @param optional nil or *GetKafkasOpts - Optional Parameters:
  * @param "Page" (optional.String) -  Page index
  * @param "Size" (optional.String) -  Number of items in each page
- * @param "OrderBy" (optional.String) -  Specifies the order by criteria. The syntax of this parameter is similar to the syntax of the `order by` clause of an SQL statement. Each query can be ordered by any of the following `kafkaRequests` fields:  * bootstrap_server_host * cloud_provider * cluster_id * created_at * href * id * instance_type * multi_az * name * organisation_id * owner * reauthentication_enabled * region * status * updated_at * version  For example, to return all Kafka instances ordered by their name, use the following syntax:  ```sql name asc ```  To return all Kafka instances ordered by their name _and_ created date, use the following syntax:  ```sql name asc, created_at asc ```  If the parameter isn't provided, or if the value is empty, then the results are ordered by name.
+ * @param "OrderBy" (optional.String) -  Specifies the order by criteria. The syntax of this parameter is similar to the syntax of the `order by` clause of an SQL statement. Each query can be ordered by any of the following `kafkaRequests` fields:  * bootstrap_server_host * admin_api_server_url * cloud_provider * cluster_id * created_at * href * id * instance_type * multi_az * name * organisation_id * owner * reauthentication_enabled * region * status * updated_at * version  For example, to return all Kafka instances ordered by their name, use the following syntax:  ```sql name asc ```  To return all Kafka instances ordered by their name _and_ created date, use the following syntax:  ```sql name asc, created_at asc ```  If the parameter isn't provided, or if the value is empty, then the results are ordered by name.
  * @param "Search" (optional.String) -  Search criteria.  The syntax of this parameter is similar to the syntax of the `where` clause of an SQL statement. Allowed fields in the search are `cloud_provider`, `name`, `owner`, `region`, and `status`. Allowed comparators are `<>`, `=`, or `LIKE`. Allowed joins are `AND` and `OR`. However, you can use a maximum of 10 joins in a search query.  Examples:  To return a Kafka instance with the name `my-kafka` and the region `aws`, use the following syntax:  ``` name = my-kafka and cloud_provider = aws ```[p-]  To return a Kafka instance with a name that starts with `my`, use the following syntax:  ``` name like my%25 ```  If the parameter isn't provided, or if the value is empty, then all the Kafka instances that the user has permission to see are returned.  Note. If the query is invalid, an error is returned.
 @return KafkaRequestList
 */

internal/kafka/internal/api/public/model_kafka_request.go

@@ -24,10 +24,12 @@ type KafkaRequest struct {
 	CloudProvider string `json:"cloud_provider,omitempty"`
 	MultiAz       bool   `json:"multi_az"`
 	// Values will be regions of specific cloud provider. For example: us-east-1 for AWS
-	Region                      string     `json:"region,omitempty"`
-	Owner                       string     `json:"owner,omitempty"`
-	Name                        string     `json:"name,omitempty"`
-	BootstrapServerHost         string     `json:"bootstrap_server_host,omitempty"`
+	Region              string `json:"region,omitempty"`
+	Owner               string `json:"owner,omitempty"`
+	Name                string `json:"name,omitempty"`
+	BootstrapServerHost string `json:"bootstrap_server_host,omitempty"`
+	// The kafka admin server url to perform kafka admin operations e.g acl management etc. The value will be available when the Kafka has been fully provisioned i.e it reaches a 'ready' state
+	AdminApiServerUrl           string     `json:"admin_api_server_url,omitempty"`
 	CreatedAt                   time.Time  `json:"created_at,omitempty"`
 	ExpiresAt                   *time.Time `json:"expires_at,omitempty"`
 	UpdatedAt                   time.Time  `json:"updated_at,omitempty"`

internal/kafka/internal/migrations/202205061532_add_admin_api_server_url.go

@@ -0,0 +1,27 @@
+package migrations
+
+import (
+	"github.com/go-gormigrate/gormigrate/v2"
+	"gorm.io/gorm"
+)
+
+// Migrations should NEVER use types from other packages. Types can change
+// and then migrations run on a _new_ database will fail or behave unexpectedly.
+// Instead of importing types, always re-create the type in the migration, as
+// is done here, even though the same type is defined in pkg/api
+
+func addAdminApiServerURL() *gormigrate.Migration {
+	type KafkaRequest struct {
+		AdminApiServerURL string `json:"admin_api_server_url" gorm:"default:''"`
+	}
+
+	return &gormigrate.Migration{
+		ID: "202205061532",
+		Migrate: func(tx *gorm.DB) error {
+			return tx.AutoMigrate(&KafkaRequest{})
+		},
+		Rollback: func(tx *gorm.DB) error {
+			return tx.Migrator().DropColumn(&KafkaRequest{}, "admin_api_server_url")
+		},
+	}
+}

internal/kafka/internal/migrations/migrations.go

@@ -77,6 +77,7 @@ var migrations = []*gormigrate.Migration{
 	addClusterServiceClientSecret(),
 	addKafkaSizeId(),
 	dropKafkaSsoClientIdAndSecret(),
+	addAdminApiServerURL(),
 }
 
 func New(dbConfig *db.DatabaseConfig) (*db.Migration, func(), error) {

internal/kafka/internal/presenters/data_plane_kafka_status.go

@@ -34,6 +34,7 @@ func ConvertDataPlaneKafkaStatus(status map[string]private.DataPlaneKafkaStatus)
 			KafkaVersion:    v.Versions.Kafka,
 			StrimziVersion:  v.Versions.Strimzi,
 			KafkaIBPVersion: v.Versions.KafkaIbp,
+			AdminServerURI:  v.AdminServerURI,
 		})
 	}
 

internal/kafka/internal/presenters/kafka.go

@@ -75,6 +75,7 @@ func PresentKafkaRequest(kafkaRequest *dbapi.KafkaRequest, config *config.KafkaC
 		MultiAz:                     kafkaRequest.MultiAZ,
 		Owner:                       kafkaRequest.Owner,
 		BootstrapServerHost:         setBootstrapServerHost(kafkaRequest.BootstrapServerHost),
+		AdminApiServerUrl:           kafkaRequest.AdminApiServerURL,
 		Status:                      kafkaRequest.Status,
 		CreatedAt:                   kafkaRequest.CreatedAt,
 		UpdatedAt:                   kafkaRequest.UpdatedAt,

internal/kafka/internal/services/data_plane_kafka.go

@@ -76,6 +76,7 @@ func (d *dataPlaneKafkaService) UpdateDataPlaneKafkaService(ctx context.Context,
 			// Store the routes (and create them) when Kafka is ready. By the time it is ready, the routes should definitely be there.
 			e = d.persistKafkaRoutes(kafka, ks, cluster)
 			if e == nil {
+				kafka.AdminApiServerURL = ks.AdminServerURI
 				e = d.setKafkaClusterReady(kafka)
 			}
 		case statusInstalling:
@@ -121,10 +122,11 @@ func (d *dataPlaneKafkaService) setKafkaClusterReady(kafka *dbapi.KafkaRequest)
 		return err
 	}
 
-	err = d.kafkaService.Updates(kafka, map[string]interface{}{"failed_reason": "", "status": constants2.KafkaRequestStatusReady.String()})
+	err = d.kafkaService.Updates(kafka, map[string]interface{}{"admin_api_server_url": kafka.AdminApiServerURL, "failed_reason": "", "status": constants2.KafkaRequestStatusReady.String()})
 	if err != nil {
-		return serviceError.NewWithCause(err.Code, err, "failed to update status %s for kafka cluster %s", constants2.KafkaRequestStatusReady, kafka.ID)
+		return serviceError.NewWithCause(err.Code, err, "failed to update kafka cluster %s", kafka.ID)
 	}
+
 	if shouldSendMetric {
 		metrics.UpdateKafkaRequestsStatusSinceCreatedMetric(constants2.KafkaRequestStatusReady, kafka.ID, kafka.ClusterID, time.Since(kafka.CreatedAt))
 		metrics.UpdateKafkaCreationDurationMetric(metrics.JobTypeKafkaCreate, time.Since(kafka.CreatedAt))