Add SQLite state store for persistent resources

README.md

@@ -34,6 +34,7 @@ DCP has knowledge of a number of environment variables that can change its behav
 | `DCP_EXTENSIONS_PATH` | Points to directory that contains DCP extensions. By default extensions are placed in the `ext` sub-directory of the directory where DCP main executable is located. |
 | `DEBUG_SESSION_PORT`, `DEBUG_SESSION_TOKEN`, and `DEBUG_SESSION_SERVER_CERTIFICATE` | These are variables that configure the endpoint for running Executables via a developer IDE/under debugger. For more information see [IDE execution specification](https://github.com/dotnet/aspire/blob/main/docs/specs/IDE-execution.md). |
 | `DCP_SESSION_FOLDER` | This variable is used for isolating multiple DCP instances running concurrently on the same machine. If set (to a valid filesystem folder), DCP process(es) will create files related to their execution in this folder: the access configuration file (kubeconfig), captured Executable/Container logs, etc. |
+| `DCP_STATE_STORE_PATH` | Overrides the path to the local SQLite state store used for DCP coordination metadata such as process records and resource leases. Resource leases are owned by the leasing DCP process identity and stale leases are cleaned up on startup. If unset, DCP uses the default per-user state store under the DCP user directory. |
 | `DCP_LOG_SOCKET` | If set to a Unix domain socket, DCP will write its execution logs to that socket instead of writing them to standard error stream (`stderr`). This allows programs that launch DCP to capture its output even if DCP is running in `--detach` mode. <br/> The `--detach` mode causes DCP to fork itself and break the parent-child relationship (and lifetime dependency) from the process that launched it, but the side effect of doing so is that the parent process loses ability to monitor DCP standard output and standard error streamd. |
 | `DCP_LOG_SESSION_ID` | If set, DCP will prepend this value to all diagnostics log names. If unset, a session ID will be calculated. The value is propagated to all child DCP processes. |
 | `DCP_DIAGNOSTICS_LOG_LEVEL` | If set, enabled DCP diagnostic logging. <br/> Can be set to `error`, `info`, or `debug`; for troubleshooting `debug` is recommended, although it results in the most verbose output. |

api/v1/container_network_types.go

@@ -7,7 +7,10 @@ package v1
 
 import (
 	"context"
+	"fmt"
+	"strings"
 
+	"github.com/microsoft/dcp/internal/statestore"
 	"github.com/microsoft/dcp/pkg/commonapi"
 	apiserver_resource "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource"
 	apiserver_resourcerest "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource/resourcerest"
@@ -107,6 +110,10 @@ func (cn *ContainerNetwork) GetGroupVersionResource() schema.GroupVersionResourc
 	}
 }
 
+func (cn *ContainerNetwork) GetLeaseKey() string {
+	return fmt.Sprintf("%s/%s", cn.GetGroupVersionResource().Resource, strings.TrimSpace(cn.Spec.NetworkName))
+}
+
 func (cn *ContainerNetwork) GetObjectMeta() *metav1.ObjectMeta {
 	return &cn.ObjectMeta
 }
@@ -214,3 +221,4 @@ var _ apiserver_resource.StatusSubResource = (*ContainerNetworkStatus)(nil)
 var _ apiserver_resourcerest.ShortNamesProvider = (*ContainerNetwork)(nil)
 var _ apiserver_resourcestrategy.Validater = (*ContainerNetwork)(nil)
 var _ apiserver_resourcestrategy.ValidateUpdater = (*ContainerNetwork)(nil)
+var _ statestore.LeasableResource = (*ContainerNetwork)(nil)

api/v1/container_types.go

@@ -34,6 +34,7 @@ import (
 	apiserver_resourcerest "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource/resourcerest"
 	apiserver_resourcestrategy "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource/resourcestrategy"
 
+	"github.com/microsoft/dcp/internal/statestore"
 	"github.com/microsoft/dcp/pkg/commonapi"
 	"github.com/microsoft/dcp/pkg/pointers"
 )
@@ -1175,6 +1176,10 @@ func (c *Container) GetGroupVersionResource() schema.GroupVersionResource {
 	}
 }
 
+func (c *Container) GetLeaseKey() string {
+	return fmt.Sprintf("%s/%s", c.GetGroupVersionResource().Resource, strings.TrimSpace(c.Spec.ContainerName))
+}
+
 func (c *Container) GetObjectMeta() *metav1.ObjectMeta {
 	return &c.ObjectMeta
 }
@@ -1550,3 +1555,4 @@ var _ apiserver_resourcestrategy.Validater = (*Container)(nil)
 var _ apiserver_resourcestrategy.ValidateUpdater = (*Container)(nil)
 var _ apiserver_resource.ObjectWithGenericSubResource = (*Container)(nil)
 var _ apiserver_resource.GenericSubResource = (*ContainerLogResource)(nil)
+var _ statestore.LeasableResource = (*Container)(nil)

api/v1/container_types_test.go

@@ -13,6 +13,24 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation/field"
 )
 
+func TestContainerGetLeaseKey(t *testing.T) {
+	t.Parallel()
+
+	container := &Container{}
+	container.Spec.ContainerName = " api "
+
+	require.Equal(t, "containers/api", container.GetLeaseKey())
+}
+
+func TestContainerNetworkGetLeaseKey(t *testing.T) {
+	t.Parallel()
+
+	network := &ContainerNetwork{}
+	network.Spec.NetworkName = " app-network "
+
+	require.Equal(t, "containernetworks/app-network", network.GetLeaseKey())
+}
+
 func TestImageLayerValidate(t *testing.T) {
 	t.Parallel()
 

api/v1/executable_types.go

@@ -7,10 +7,16 @@ package v1
 
 import (
 	"context"
+	"encoding/gob"
+	"errors"
 	"fmt"
+	"hash/fnv"
 	stdmaps "maps"
+	"os"
 	stdslices "slices"
+	"strings"
 
+	"github.com/joho/godotenv"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -24,7 +30,11 @@ import (
 	apiserver_resourcerest "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource/resourcerest"
 	apiserver_resourcestrategy "github.com/tilt-dev/tilt-apiserver/pkg/server/builder/resource/resourcestrategy"
 
+	"github.com/microsoft/dcp/internal/statestore"
 	"github.com/microsoft/dcp/pkg/commonapi"
+	usvc_maps "github.com/microsoft/dcp/pkg/maps"
+	"github.com/microsoft/dcp/pkg/osutil"
+	"github.com/microsoft/dcp/pkg/pointers"
 	"github.com/microsoft/dcp/pkg/slices"
 )
 
@@ -250,10 +260,21 @@ type ExecutableSpec struct {
 	// Controls behavior of environment variables inherited from the controller process.
 	AmbientEnvironment AmbientEnvironment `json:"ambientEnvironment,omitempty"`
 
+	// Should the controller attempt to start the Executable?
+	// +kubebuilder:default:=true
+	Start *bool `json:"start,omitempty"`
+
 	// Should the controller attempt to stop the Executable
 	// +kubebuilder:default:=false
 	Stop bool `json:"stop,omitempty"`
 
+	// Should this Executable be created and persisted between DCP runs?
+	Persistent bool `json:"persistent,omitempty"`
+
+	// Optional key used to identify if an existing persistent Executable process should be reused.
+	// If not set, the controller will calculate a key based on a hash of specific fields in the ExecutableSpec.
+	LifecycleKey string `json:"lifecycleKey,omitempty"`
+
 	// Health probe configuration for the Executable
 	// +listType=atomic
 	HealthProbes []HealthProbe `json:"healthProbes,omitempty"`
@@ -296,10 +317,22 @@ func (es ExecutableSpec) Equal(other ExecutableSpec) bool {
 		return false
 	}
 
+	if pointers.GetValueOrDefault(es.Start, true) != pointers.GetValueOrDefault(other.Start, true) {
+		return false
+	}
+
 	if es.Stop != other.Stop {
 		return false
 	}
 
+	if es.Persistent != other.Persistent {
+		return false
+	}
+
+	if es.LifecycleKey != other.LifecycleKey {
+		return false
+	}
+
 	if len(es.HealthProbes) != len(other.HealthProbes) {
 		return false
 	}
@@ -317,6 +350,159 @@ func (es ExecutableSpec) Equal(other ExecutableSpec) bool {
 	return true
 }
 
+func (es *ExecutableSpec) GetLifecycleKey() (string, bool, error) {
+	if es.LifecycleKey != "" {
+		return es.LifecycleKey, false, nil
+	}
+
+	fnvHash := fnv.New128()
+	encoder := gob.NewEncoder(fnvHash)
+
+	var hashErr error
+	hashErr = errors.Join(hashErr, encoder.Encode(es.ExecutablePath))
+	hashErr = errors.Join(hashErr, encoder.Encode(es.WorkingDirectory))
+	hashErr = errors.Join(hashErr, encoder.Encode(string(es.ExecutionType)))
+	hashErr = errors.Join(hashErr, encoder.Encode(string(es.AmbientEnvironment.Behavior)))
+	hashErr = errors.Join(hashErr, encoder.Encode(es.Args))
+
+	if len(es.Env) > 0 {
+		sortedEnv := stdslices.Clone(es.Env)
+		stdslices.SortFunc(sortedEnv, func(e1, e2 EnvVar) int {
+			return strings.Compare(e1.Name, e2.Name)
+		})
+
+		for i := range sortedEnv {
+			hashErr = errors.Join(hashErr, encoder.Encode(sortedEnv[i]))
+		}
+	}
+
+	if len(es.EnvFiles) > 0 {
+		sortedEnvFiles := stdslices.Clone(es.EnvFiles)
+		stdslices.Sort(sortedEnvFiles)
+
+		for i := range sortedEnvFiles {
+			envFileContents, envFileReadErr := os.ReadFile(sortedEnvFiles[i])
+			if envFileReadErr != nil {
+				hashErr = errors.Join(hashErr, envFileReadErr)
+			} else {
+				hashErr = errors.Join(hashErr, encoder.Encode(envFileContents))
+			}
+		}
+	}
+
+	if es.PemCertificates != nil {
+		sortedPemCertificates := stdslices.Clone(es.PemCertificates.Certificates)
+		stdslices.SortFunc(sortedPemCertificates, func(c1, c2 PemCertificate) int {
+			return strings.Compare(c1.Thumbprint, c2.Thumbprint)
+		})
+
+		for i := range sortedPemCertificates {
+			hashErr = errors.Join(hashErr, encoder.Encode(sortedPemCertificates[i]))
+		}
+		hashErr = errors.Join(hashErr, encoder.Encode(es.PemCertificates.ContinueOnError))
+	}
+
+	lifecycleKey := fmt.Sprintf("%x", fnvHash.Sum(nil))
+	return lifecycleKey, true, hashErr
+}
+
+func (e *Executable) GetLifecycleKey() (string, bool, error) {
+	lifecycleSpec, lifecycleSpecErr := e.EffectiveLifecycleSpec()
+	if lifecycleSpecErr != nil {
+		return "", false, lifecycleSpecErr
+	}
+	return lifecycleSpec.GetLifecycleKey()
+}
+
+func (e *Executable) EffectiveLifecycleSpec() (ExecutableSpec, error) {
+	lifecycleSpec := *e.Spec.DeepCopy()
+	effectiveArgs, effectiveArgsErr := effectiveLifecycleArgs(e)
+	if effectiveArgsErr != nil {
+		return ExecutableSpec{}, effectiveArgsErr
+	}
+	explicitEffectiveEnv, explicitEffectiveEnvErr := explicitEffectiveLifecycleEnv(e)
+	if explicitEffectiveEnvErr != nil {
+		return ExecutableSpec{}, explicitEffectiveEnvErr
+	}
+	lifecycleSpec.Args = effectiveArgs
+	lifecycleSpec.Env = explicitEffectiveEnv
+	lifecycleSpec.EnvFiles = nil
+	return lifecycleSpec, nil
+}
+
+func effectiveLifecycleArgs(e *Executable) ([]string, error) {
+	if len(e.Status.EffectiveArgs) == 0 {
+		if len(e.Spec.Args) > 0 {
+			return nil, fmt.Errorf("executable lifecycle key cannot be calculated before effective arguments are computed")
+		}
+		return nil, nil
+	}
+	return stdslices.Clone(e.Status.EffectiveArgs), nil
+}
+
+func explicitEffectiveLifecycleEnv(e *Executable) ([]EnvVar, error) {
+	explicitNames := explicitLifecycleEnvNames(e)
+	if explicitNames.Len() == 0 {
+		return nil, nil
+	}
+
+	effectiveEnvByName := lifecycleEnvMap()
+	for _, envVar := range e.Status.EffectiveEnv {
+		effectiveEnvByName.Set(envVar.Name, envVar.Value)
+	}
+	if effectiveEnvByName.Len() == 0 {
+		return nil, fmt.Errorf("executable lifecycle key cannot be calculated before effective environment is computed")
+	}
+
+	explicitEffectiveEnv := make([]EnvVar, 0, explicitNames.Len())
+	for nameKey, name := range explicitNames.Data() {
+		value, found := effectiveEnvByName.Get(nameKey)
+		if !found {
+			continue
+		}
+		explicitEffectiveEnv = append(explicitEffectiveEnv, EnvVar{Name: name, Value: value})
+	}
+	stdslices.SortFunc(explicitEffectiveEnv, func(e1, e2 EnvVar) int {
+		return strings.Compare(lifecycleEnvKey(e1.Name), lifecycleEnvKey(e2.Name))
+	})
+	return explicitEffectiveEnv, nil
+}
+
+func explicitLifecycleEnvNames(e *Executable) usvc_maps.StringKeyMap[string] {
+	explicitNames := lifecycleEnvMap()
+	addExplicitName := func(name string) {
+		if name != "" {
+			explicitNames.Set(name, name)
+		}
+	}
+
+	if len(e.Spec.EnvFiles) > 0 {
+		if fileEnv, readErr := godotenv.Read(e.Spec.EnvFiles...); readErr == nil {
+			for name := range fileEnv {
+				addExplicitName(name)
+			}
+		}
+	}
+	for _, envVar := range e.Spec.Env {
+		addExplicitName(envVar.Name)
+	}
+	return explicitNames
+}
+
+func lifecycleEnvMap() usvc_maps.StringKeyMap[string] {
+	if osutil.IsWindows() {
+		return usvc_maps.NewStringKeyMap[string](usvc_maps.StringMapModeCaseInsensitive)
+	}
+	return usvc_maps.NewStringKeyMap[string](usvc_maps.StringMapModeCaseSensitive)
+}
+
+func lifecycleEnvKey(name string) string {
+	if osutil.IsWindows() {
+		return strings.ToUpper(name)
+	}
+	return name
+}
+
 func (es ExecutableSpec) Validate(specPath *field.Path) field.ErrorList {
 	errorList := field.ErrorList{}
 
@@ -348,6 +534,17 @@ func (es ExecutableSpec) Validate(specPath *field.Path) field.ErrorList {
 		errorList = append(errorList, field.Invalid(specPath.Child("ambientEnvironment", "behavior"), es.AmbientEnvironment.Behavior, "Ambient environment behavior must be either Inherit or DoNotInherit."))
 	}
 
+	effectiveExecutionType := es.ExecutionType
+	if effectiveExecutionType == "" {
+		effectiveExecutionType = ExecutionTypeProcess
+	}
+	if es.Persistent && effectiveExecutionType != ExecutionTypeProcess {
+		errorList = append(errorList, field.Invalid(specPath.Child("persistent"), es.Persistent, "Persistent Executables only support Process execution type."))
+	}
+	if es.Persistent && len(es.FallbackExecutionTypes) > 0 {
+		errorList = append(errorList, field.Invalid(specPath.Child("fallbackExecutionTypes"), es.FallbackExecutionTypes, "Persistent Executables cannot use fallback execution types."))
+	}
+
 	healthProbesPath := specPath.Child("healthProbes")
 	for i, probe := range es.HealthProbes {
 		errorList = append(errorList, probe.Validate(healthProbesPath.Index(i))...)
@@ -453,6 +650,10 @@ func (e *Executable) GetResourceId() string {
 	return fmt.Sprintf("executable-%s", e.UID)
 }
 
+func (e *Executable) GetLeaseKey() string {
+	return e.NamespacedName().String()
+}
+
 func (e *Executable) GetGroupVersionResource() schema.GroupVersionResource {
 	return schema.GroupVersionResource{
 		Group:    GroupVersion.Group,
@@ -517,10 +718,18 @@ func (e *Executable) ValidateUpdate(ctx context.Context, obj runtime.Object) fie
 	errorList := field.ErrorList{}
 
 	oldExe := obj.(*Executable)
+	if (oldExe.Spec.Start == nil || *oldExe.Spec.Start) && (e.Spec.Start != nil && !*e.Spec.Start) {
+		errorList = append(errorList, field.Forbidden(field.NewPath("spec", "start"), "Cannot set start to false after Executable creation."))
+	}
+
 	if oldExe.Spec.Stop && e.Spec.Stop != oldExe.Spec.Stop {
 		errorList = append(errorList, field.Forbidden(field.NewPath("spec", "stop"), "Cannot unset stop property once it is set."))
 	}
 
+	if oldExe.Spec.Persistent != e.Spec.Persistent {
+		errorList = append(errorList, field.Forbidden(field.NewPath("spec", "persistent"), "persistent cannot be changed"))
+	}
+
 	if oldExe.Spec.AmbientEnvironment.Behavior != e.Spec.AmbientEnvironment.Behavior {
 		errorList = append(errorList, field.Forbidden(field.NewPath("spec", "ambientEnvironment", "behavior"), "Cannot change ambient environment behavior once it is set."))
 	}
@@ -547,6 +756,10 @@ func (e *Executable) Done() bool {
 	return !e.Status.FinishTimestamp.IsZero()
 }
 
+func (e *Executable) ShouldStart() bool {
+	return e.Spec.Start == nil || *e.Spec.Start
+}
+
 func (*Executable) GenericSubResources() []apiserver_resource.GenericSubResource {
 	return []apiserver_resource.GenericSubResource{
 		&ExecutableLogResource{},
@@ -630,3 +843,4 @@ var _ apiserver_resourcestrategy.ValidateUpdater = (*Executable)(nil)
 var _ apiserver_resource.ObjectWithGenericSubResource = (*Executable)(nil)
 var _ apiserver_resource.GenericSubResource = (*ExecutableLogResource)(nil)
 var _ StdIoStreamableResource = (*Executable)(nil)
+var _ statestore.LeasableResource = (*Executable)(nil)