✨ Support multiple provider in separate containers. (#606)

Support dynamic selection of multiple providers in separate containers.

closes #599 

Summary by package:
- /addon
- add support for addons to fetch the _Addon_ using the binding through
the addon (adapter) package.
  - add environment variables injector.  replaces $(var) with values.
- /api
  - addon 
    - support changes in addon and extension CRs. 
    - included (new) extensions in _Addon_ resource.
  - task
    - removed unused fields: variant, purged. 
    - support new addon and extensions fields.
- resolve k8s resources referenced by task: addon, extension, task,
priority for validation and default priority.
   - taskgroup - support changes in task.
   - delegate CRUD (including cancel) to the task manager. This ensure:
     - tasks in flight are not updated by multiple threads.
- delegates complexities on what can be updated when to the task manger.
 - /binding - add support for reading Get, List addons.
 - generated/crd - result of: (make generate && make manifests)
- /k8s/api -
   - addon - updated to support dynamic addon selection.
   - task - added to support dynamic addon selection.
   - extension - added to support extensions with dynamic selection.
- /migration/v13 - core.go updated.
   - Task.kind added.
   - Task.Extensions added. 
- Task.Attached added to support attached pod definition and container
logs.
- Task.Policy structured _TaskPolicy_. which includes preemption
policies. Structured improved the api.
- /model - ref migration v13.
- /reaper
  -  reap files attached to tasks.
  - report released event.
- /settings - add hub setting for location of /shared (emptydir).
- /task - support:
- converted task models to use json serializer (while adding other json
fields).
- multi-container task pods which ensured extension container
termination.
  - both static and dynamic addon seletion.
  - both static and dynamic extension selection. 
  - task (kinds)
- priority escalation based on _task-kind_ dependencies. Prevents
priority inversions.
  - task preemption based on priority.
- enhanced ordering based on _task-kind_ dependency. adds task Postponed
rule.
- attach pod.yaml (including events) and container logs to the task on
pod completion.
  - support resource quota.
- add environment variable injection _(mainly to support automatic port
assignment for providers)._
- add ${seq:<pool>} macro _(mainly to support automatic port assignment
for providers)._
  - add task event reporting.
  - refactored the manager:
    - task state simplified: pending state removed (reported by event).
    -  consolidate errors into error.go and add _soft_ error concept.
- Manager.runReady() into separate methods for better decomposition.
each loop through and acts on the list.
- Manager.updateRunning() into separate methods for better functional
decomposition.
- fetch all k8s resources at once instead of hammering the api-server.
---

json serialization: added a custom json (model) serializer. This was
needed because we need the serializer to
handle the map[any]any problem with binding yaml. Removed
api/base.go/strMap() and moved it to the new serializer. Only being used
by tasks.

---

Added/Updated CRDs:
- Addon (updated)
- Extension
- Task

---------

Signed-off-by: Jeff Ortel <jortel@redhat.com>

addon/adapter.go

@@ -21,6 +21,14 @@ var (
 	Log      = logr.WithName("addon")
 )
 
+// Environment.
+const (
+	EnvSharedDir = settings.EnvSharedPath
+	EnvCacheDir  = settings.EnvCachePath
+	EnvToken     = settings.EnvHubToken
+	EnvTask      = settings.EnvTask
+)
+
 // Addon An addon adapter configured for a task execution.
 var Addon *Adapter
 

addon/injector.go

@@ -0,0 +1,76 @@
+package addon
+
+import (
+	"encoding/json"
+	"os"
+	"regexp"
+	"strings"
+
+	"github.com/konveyor/tackle2-hub/api"
+	"github.com/konveyor/tackle2-hub/task"
+)
+
+var (
+	EnvRegex = regexp.MustCompile(`(\$\()([^)]+)(\))`)
+)
+
+// EnvInjector inject key into extension metadata.
+type EnvInjector struct {
+	env  map[string]string
+	dict map[string]string
+}
+
+// Inject inject into extension metadata.
+func (r *EnvInjector) Inject(extension *api.Extension) {
+	r.buildEnv(extension)
+	mp := make(map[string]any)
+	b, _ := json.Marshal(extension.Metadata)
+	_ = json.Unmarshal(b, &mp)
+	mp = r.inject(mp).(map[string]any)
+	extension.Metadata = mp
+}
+
+// buildEnv builds the extension `env`.
+func (r *EnvInjector) buildEnv(extension *api.Extension) {
+	r.env = make(map[string]string)
+	for _, env := range extension.Container.Env {
+		key := task.ExtEnv(extension.Name, env.Name)
+		r.env[env.Name] = os.Getenv(key)
+	}
+}
+
+// inject replaces both `dict` keys and `env` environment
+// variables referenced in metadata.
+func (r *EnvInjector) inject(in any) (out any) {
+	switch node := in.(type) {
+	case map[string]any:
+		for k, v := range node {
+			node[k] = r.inject(v)
+		}
+		out = node
+	case []any:
+		var injected []any
+		for _, n := range node {
+			injected = append(
+				injected,
+				r.inject(n))
+		}
+		out = injected
+	case string:
+		for {
+			match := EnvRegex.FindStringSubmatch(node)
+			if len(match) < 3 {
+				break
+			}
+			node = strings.Replace(
+				node,
+				match[0],
+				r.env[match[2]],
+				-1)
+		}
+		out = node
+	default:
+		out = node
+	}
+	return
+}

addon/task.go

@@ -42,9 +42,48 @@ func (h *Task) Application() (r *api.Application, err error) {
 	return
 }
 
+// Addon returns the addon associated with the task.
+// The extensions are filtered to include those specified in the task.
+// inject: perform injection.
+func (h *Task) Addon(inject bool) (r *api.Addon, err error) {
+	name := h.task.Addon
+	if name == "" {
+		err = &NotFound{}
+		return
+	}
+	r, err = h.richClient.Addon.Get(name)
+	if err != nil {
+		return
+	}
+	// filter
+	included := map[string]int{}
+	for _, name := range h.task.Extensions {
+		included[name] = 0
+	}
+	var extensions []api.Extension
+	for i := range r.Extensions {
+		extension := r.Extensions[i]
+		if _, found := included[extension.Name]; found {
+			extensions = append(
+				extensions,
+				extension)
+		}
+	}
+	r.Extensions = extensions
+	// inject
+	if inject {
+		for i := range r.Extensions {
+			extension := &r.Extensions[i]
+			injector := EnvInjector{}
+			injector.Inject(extension)
+		}
+	}
+	return
+}
+
 // Data returns the addon data.
-func (h *Task) Data() (d map[string]interface{}) {
-	d = h.task.Data.(map[string]interface{})
+func (h *Task) Data() (d api.Map) {
+	d = h.task.Data
 	return
 }
 
@@ -55,11 +94,6 @@ func (h *Task) DataWith(object interface{}) (err error) {
 	return
 }
 
-// Variant returns the task variant.
-func (h *Task) Variant() string {
-	return h.task.Variant
-}
-
 // Started report addon started.
 func (h *Task) Started() {
 	h.deleteReport()
@@ -169,10 +203,8 @@ func (h *Task) AttachAt(f *api.File, activity int) {
 		h.report.Attached,
 		api.Attachment{
 			Activity: activity,
-			Ref: api.Ref{
-				ID:   f.ID,
-				Name: f.Name,
-			},
+			ID:       f.ID,
+			Name:     f.Name,
 		})
 	h.pushReport()
 	return
@@ -215,7 +247,7 @@ func (h *Task) Bucket() (b *binding.BucketContent) {
 }
 
 // Result report addon result.
-func (h *Task) Result(object interface{}) {
+func (h *Task) Result(object api.Map) {
 	h.report.Result = object
 	h.pushReport()
 	Log.Info("Addon reported: result.")

api/addon.go

@@ -2,10 +2,12 @@ package api
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 
 	"github.com/gin-gonic/gin"
 	crd "github.com/konveyor/tackle2-hub/k8s/api/tackle/v1alpha1"
+	core "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	k8s "sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -57,8 +59,19 @@ func (h AddonHandler) Get(ctx *gin.Context) {
 			return
 		}
 	}
+	extensions := &crd.ExtensionList{}
+	err = h.Client(ctx).List(
+		context.TODO(),
+		extensions,
+		&k8s.ListOptions{
+			Namespace: Settings.Namespace,
+		})
+	if err != nil {
+		_ = ctx.Error(err)
+		return
+	}
 	r := Addon{}
-	r.With(addon)
+	r.With(addon, extensions.Items...)
 
 	h.Respond(ctx, http.StatusOK, r)
 }
@@ -94,12 +107,43 @@ func (h AddonHandler) List(ctx *gin.Context) {
 
 // Addon REST resource.
 type Addon struct {
-	Name  string `json:"name"`
-	Image string `json:"image"`
+	Name       string         `json:"name"`
+	Container  core.Container `json:"container"`
+	Extensions []Extension    `json:"extensions,omitempty"`
+	Metadata   any            `json:"metadata,omitempty"`
 }
 
 // With model.
-func (r *Addon) With(m *crd.Addon) {
+func (r *Addon) With(m *crd.Addon, extensions ...crd.Extension) {
 	r.Name = m.Name
-	r.Image = m.Spec.Image
+	r.Container = m.Spec.Container
+	if m.Spec.Metadata.Raw != nil {
+		_ = json.Unmarshal(m.Spec.Metadata.Raw, &r.Metadata)
+	}
+	for i := range extensions {
+		extension := Extension{}
+		extension.With(&extensions[i])
+		r.Extensions = append(
+			r.Extensions,
+			extension)
+	}
+}
+
+// Extension REST resource.
+type Extension struct {
+	Name         string         `json:"name"`
+	Addon        string         `json:"addon"`
+	Capabilities []string       `json:"capabilities,omitempty"`
+	Container    core.Container `json:"container"`
+	Metadata     any            `json:"metadata,omitempty"`
+}
+
+// With model.
+func (r *Extension) With(m *crd.Extension) {
+	r.Name = m.Name
+	r.Addon = m.Spec.Addon
+	r.Container = m.Spec.Container
+	if m.Spec.Metadata.Raw != nil {
+		_ = json.Unmarshal(m.Spec.Metadata.Raw, &r.Metadata)
+	}
 }

api/base.go

@@ -461,19 +461,3 @@ func (r *Cursor) pageLimited() (b bool) {
 	b = r.Index > int64(r.Limit)
 	return
 }
-
-// StrMap returns a map[string]any.
-// The YAML decoder can produce map[any]any which is not valid for json.
-// Converts map[any]any to map[string]any as needed.
-func StrMap(in any) (out any) {
-	out = in
-	if d, cast := in.(map[any]any); cast {
-		mp := make(map[string]any)
-		for k, v := range d {
-			s := fmt.Sprintf("%v", k)
-			mp[s] = StrMap(v)
-		}
-		out = mp
-	}
-	return
-}

api/context.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/konveyor/tackle2-hub/auth"
+	tasking "github.com/konveyor/tackle2-hub/task"
 	"gorm.io/gorm"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -22,6 +23,8 @@ type Context struct {
 	Client client.Client
 	// Response
 	Response Response
+	// Task manager.
+	TaskManager *tasking.Manager
 }
 
 // Response values.

api/error.go

@@ -10,6 +10,7 @@ import (
 	"github.com/konveyor/tackle2-hub/api/filter"
 	"github.com/konveyor/tackle2-hub/api/sort"
 	"github.com/konveyor/tackle2-hub/model"
+	tasking "github.com/konveyor/tackle2-hub/task"
 	"github.com/mattn/go-sqlite3"
 	"gorm.io/gorm"
 )
@@ -173,6 +174,15 @@ func ErrorHandler() gin.HandlerFunc {
 			return
 		}
 
+		if errors.Is(err, &tasking.BadRequest{}) {
+			rtx.Respond(
+				http.StatusBadRequest,
+				gin.H{
+					"error": err.Error(),
+				})
+			return
+		}
+
 		rtx.Respond(
 			http.StatusInternalServerError,
 			gin.H{