Plugin architectured piped (pipedv1) supports deployment for applications

[khanhtc1202]

What would you like to be added:

This issue is for managing the process of adopting plugin architecture to piped. The ideal result is to make all platform providers' logic independent from the piped core logic (the controller package).

Tasks pool break down 💪

• [x] Revise the plugin planner interface <- @khanhtc1202 @Warashi
• [ ] Implement piped plugin grpc service <- @khanhtc1202
• [x] Implement piped plugin grpc secret decryption endpoint <- @khanhtc1202
• [x] Implement piped planner which calls to the plugin deployment service <- @khanhtc1202
• [ ] Implement k8s plugin: PlannerService interface implementation <- @Warashi
• [ ] Update model.PipelineStage model:
  • [x] mark Predefined and Visible as deprecated
  • [x] add rollback field
  • [ ] add timeout field

Why is this needed:

[khanhtc1202]

As the current stage, the interface for the plugin of the piped as below We have 2 services: The PlannerService

// PlannerService defines the public APIs for remote planners.
service PlannerService {
    // BuildPlan builds plan for the given deployment.
    rpc BuildPlan(BuildPlanRequest) returns (BuildPlanResponse) {}
}

message BuildPlanRequest {
    string working_dir = 1 [(validate.rules).string.min_len = 1];
    // Last successful commit hash and config file name.
    // Use to build deployment source object for last successful deployment.
    string last_successful_commit_hash = 2;
    string last_successful_config_file_name = 3;
    // The configuration of the piped that handles the deployment.
    bytes piped_config = 4 [(validate.rules).bytes.min_len = 1];
    // The deployment to build a plan for.
    model.Deployment deployment = 5 [(validate.rules).message.required = true];
}

message BuildPlanResponse {
    // The built deployment plan.
    DeploymentPlan plan = 1;
}

message DeploymentPlan {
    model.SyncStrategy sync_strategy = 1;
    // Text summary of planned deployment.
    string summary = 2;
    repeated model.ArtifactVersion versions = 3;
    repeated model.PipelineStage stages = 4;
}

And, the ExecutorService

service ExecutorService {
    // Execute executes the given stage of the deployment plan.
    rpc ExecuteStage(ExecuteStageRequest) returns (stream ExecuteStageResponse) {}
}

message ExecuteStageRequest {
    model.PipelineStage stage = 1 [(validate.rules).message.required = true];
    bytes stage_config = 2 [(validate.rules).bytes.min_len = 1];
    bytes piped_config = 3 [(validate.rules).bytes.min_len = 1];
    model.Deployment deployment = 4 [(validate.rules).message.required = true];
}

message ExecuteStageResponse {
    model.StageStatus status = 1;
    string log = 2;
}

ref: https://github.com/pipe-cd/pipecd/blob/master/pkg/plugin/api/v1alpha1/platform/api.proto

[khanhtc1202]

Latest update of the PlannerService interface

// PlannerService defines the public APIs for remote planners.
service PlannerService {
    // DetermineStrategy determines which strategy should be used for the given deployment.
    rpc DetermineStrategy(DetermineStrategyRequest) returns (DetermineStrategyResponse) {}
    // QuickSyncPlan builds plan for the given deployment using quick sync strategy.
    rpc QuickSyncPlan(QuickSyncPlanRequest) returns (QuickSyncPlanResponse) {}
    // PipelineSyncPlan builds plan for the given deployment using pipeline sync strategy.
    rpc PipelineSyncPlan(PipelineSyncPlanRequest) returns (PipelineSyncPlanResponse) {}
}

message DetermineStrategyRequest {
    PlanPluginInput input = 1 [(validate.rules).message.required = true];
}

message DetermineStrategyResponse {
    // The determined sync strategy.
    model.SyncStrategy sync_strategy = 1;
    // Text summary of the determined strategy.
    string summary = 2;
}

message QuickSyncPlanRequest {
    PlanPluginInput input = 1 [(validate.rules).message.required = true];
}

message QuickSyncPlanResponse {
    // Stages of deployment pipeline under quick sync strategy.
    repeated model.PipelineStage stages = 1;
}

message PipelineSyncPlanRequest {
    PlanPluginInput input = 1 [(validate.rules).message.required = true];
}

message PipelineSyncPlanResponse {
    // Stages of deployment pipeline under pipeline sync strategy.
    repeated model.PipelineStage stages = 1;
}

message PlanPluginInput {
    // The deployment to build a plan for.
    model.Deployment deployment = 1 [(validate.rules).message.required = true];
    // The remote URL of the deployment source, where plugin can find the deployments sources (manifests).
    string source_remote_url = 2 [(validate.rules).string.min_len = 1];
    // Last successful commit hash and config file name.
    // Use to build deployment source object for last successful deployment.
    string last_successful_commit_hash = 3;
    string last_successful_config_file_name = 4;
    // The configuration of plugin that handles the deployment.
    bytes plugin_config = 5;
}

[khanhtc1202]

Note: The stage.requires use Id of the previous stage (instead of the stage name of previous stage) and there is only the Stage UI render logic which is depended on that ref: https://github.com/pipe-cd/pipecd/blob/master/web/src/components/deployments-detail-page/pipeline/index.tsx#L59-L80

const createStagesForRendering = (
  deployment: Deployment.AsObject | undefined
): Stage[][] => {
  if (!deployment) {
    return [];
  }

  const stages: Stage[][] = [];
  const visibleStages = deployment.stagesList.filter((stage) => stage.visible);

  stages[0] = visibleStages.filter((stage) => stage.requiresList.length === 0);

  let index = 0;
  while (stages[index].length > 0) {
    const previousIds = stages[index].map((stage) => stage.id);
    index++;
    stages[index] = visibleStages.filter((stage) =>
      stage.requiresList.some((id) => previousIds.includes(id))
    );
  }
  return stages;
};

Note: Can't use the stage name as the alternative option for the stage id, since there could be stage name reused in the same pipeline (for example SCRIPT_RUN, WAIT, etc)

[khanhtc1202]

Note: The current stage to build model and its notes

stage := &model.PipelineStage{
    Id:         id, // Stage config setting or defaulted by index from plugin (not usable since index is not shared between plugins) -> buildable
    Name:       s.Name.String(), // Only get from plugin
    Desc:       s.Desc, // Stage config setting
    Index:      int32(i), // Unable to use since index is not shared between plugins (important)
    Predefined: false, // Only get from plugin
    Visible:    true, // Only get from plugin
    Status:     model.StageStatus_STAGE_NOT_STARTED_YET, // Always NOT_STARTED_YET
    Metadata:   planner.MakeInitialStageMetadata(s), // Piped creates metadata store while plugin create the init metadata
    CreatedAt:  now.Unix(), // Always current time
    UpdatedAt:  now.Unix(), // Always current time
}

For the index let send it from the piped (as it has all the stages in order and can build the exact stage index)

[khanhtc1202]

Reply: https://github.com/pipe-cd/pipecd/issues/4980#issuecomment-2268492287 💡 We will build the stages requires on the piped instead of using anything returned from the plugins since we have to wait until that time to have all stages' id.

[khanhtc1202]

Update: cc @Warashi ref: #5237 We're moving the application config marshal/unmarshal for piped from v0 to v1, and we realized it could be easier to make it by separating the config for pipedv0 and pipedv1. So we introduced the configv1 package, which is basically a copy of the config package with less type (due to the platform's specified type being moved to plugins). Will consider how to keep changes between config and configv1 is notable later 👀

[khanhtc1202]

Note: For current rollback stages, they all have the same requires, which is the last failed/run stage; thus all rollback will be triggered to run at one (at the same time) (cc @ffjlabo to confirm later)

[ffjlabo]

Note:

• Check this later https://github.com/pipe-cd/pipecd/issues/3086 cc @khanhtc1202

[khanhtc1202]

Note: Current executor input spec

input := executor.Input{
        Stage:               &ps, // pass this
        StageConfig:         stageConfig, // pass this
        Deployment:          s.deployment, // pass this
        Application:         app, // redundant, use Deployment's fields instead
        PipedConfig:         s.pipedConfig, // pass plugin config
        TargetDSP:           s.targetDSP, // pass this as targetDS
        RunningDSP:          s.runningDSP, // pass this as runningDS
        GitClient:           s.gitClient, // Should not pass this
        CommandLister:       cmdLister, // Shared between stages, need controlplane apiClient, should be served via piped grpc
        LogPersister:        lp, // Should pass to plugin? based on the logging method
        MetadataStore:       s.metadataStore, // Shared between stages, need controlplane apiClient, should be served via piped grpc
        AppManifestsCache:   s.appManifestsCache, // only for k8s manifest cache, should be placed in plugin
        AnalysisResultStore: aStore, // Only for analysis plugin, need controlplane apiClient, should be served via piped grpc
        AppLiveResourceLister: alrLister, // Only for k8s, should be placed in plugin
        Logger:              s.logger, // Should pass to plugin
        Notifier:            s.notifier, // Only for wait-approval, should be served via piped gprc
    }

[khanhtc1202]

Based on the above https://github.com/pipe-cd/pipecd/issues/4980#issuecomment-2452071105 input, the PluginService of piped should contains

service PluginService {
   ...
    // FetchStageCommands fetches the stage commands.
    // Plugin can use this to fetch the commands all stage commands of handling deployment.
    rpc FetchStageCommands(FetchStageCommandsRequest) returns (FetchStageCommandsResponse) {}
    // SendStageNotification sends the notification of the stage using piped notifier.
    rpc SendStageNotification(SendStageNotificationRequest) returns (SendStageNotificationResponse) {}
}

TBD:

• MetadataStore service, the current MetadataStore interface is quite complicated and had used widely across stages (1)
• AnalysisResultStore service, this be used only for analysis stage plugin, but use apiClient to connect to controlplane (store value to filestore) (2)

ref: (1) https://github.com/pipe-cd/pipecd/blob/master/pkg/app/pipedv1/metadatastore/store.go#L46-L49 (2)

type AnalysisResultStore interface {
    GetLatestAnalysisResult(ctx context.Context) (*model.AnalysisResult, error)
    PutLatestAnalysisResult(ctx context.Context, analysisResult *model.AnalysisResult) error
}