Implement caching workflow

[PhilippeMoussalli]

Related to #317
First Review PRs #318 #320
Design discussions #292

PR that implements the general component caching mechanism described in

Few notes:

• There is no need to specify previous run-id when caching since we're checking for existing executions of similar definition based on the manifest and the cache key
• Created a Metadata class since the metadata of the manifest had different schema between both the local and remote runner
• If a component is cached, it skips the execution entirely and just fetches the existing output manifest (based on the matching cache key). For kubeflow, we still have to write that manifest as an output artifact that will be read by the next component (artifact) but we don't write anything to the base oath (custom_artifact)
• Currently implemented only for the remote runner (kfp) and enabled by default, can be disabled by setting Pipeline(cache_disabled = True). Disabled for remote runner since there is no straightforward method of estimating digests in that workflow (see Add method to estimate caching key #318)
  *If one component is not cached, then all subsequent components must be re-run/executed

Things to improve in later PRs:

1. Decide whether we should still estimating the cache during compilation or execution (right now it's done at compilation time to enable defining new arguments but we might miss the fact that some image digests are updated):

A) Compilation

• Right now, the component has to be run and produce the output manifest but we already know beforehand if this component should be executed or not. Ideally we would omit execution of component altogether to make pipelines run faster and avoid pod initialization waiting time. This can be done by either:
  • Implementing conditional statement in KFP link.
  • Assigning the default node pool to the cached component in case specified otherwise (avoid initialization of expensive nodes since the component just has to write the manifest)

B) Runtime

• Find a method to estimate the digest from within the component itself (see discussion at Add method to estimate caching key #318), this solution will require all component to be run since the component itself has to decide whether it can be cached or not when it runs
• Alternatively we can have one custom component at the beginning of each pipeline runner (initializer component), that decides which components should be run. The advantage here is that we only include all the required packages for estimating digests in one docker images and not all of them. See this for more details on how image digests can be retrieved.

2. Implement the caching for the local runner

[PhilippeMoussalli]

Closed in favor of #387