Add dataproc handler

[MattWellie]

Context: for the seqera trial work the latest workflow has a few files chucked in the workflow bin here. One of these is a metamist registration script which was already ported to cpg-utils as #194. Another is a Dataproc manager. This is for the latter.

---

This takes the single-use Dataproc handler from the test repo, makes it generic, and exposes it as a context manager class. Instead of being dialled into a single use case, with one set of scripts, inputs, params etc., this tries to make those behaviours configurable. It's an alternative to the current cpg_utils.dataproc module which uses hailctl to mediate all the dataproc interactions... That's fine, but we probably don't want to be using Hail to wean ourselves off Hail:

• adds a HailDataprocCluster class
• this contains a lot of rational values for default machine types, TTL, polling intervals, but it can all be altered
• the cluster class creates a cluster, handles teardown, methods to facilitate staging physical files in the nominated temp/staging bucket attached to the cluster, allows for extra scripts/calls to be added to an existing cluster
• various helper methods to sanitise labels and cluster names, to match the google max-length and content restrictions (lower alphanumeric/hyphen, starts with a letter, labels max 63 chars, dp cluster max 52 chars)
• the log retrieval mechanic has been revised slightly - instead of always retrieving the whole log and printing with an offset, it retrieves with an offset to minimise data transfer (thanks Fable)
• list of packages to be installed at runtime inside the dataproc cluster is configurable, defaulting to cpg-utils (could be fully blank? I also had some weird situations where hail needed to be added to PKGS, Hail's own dataproc startup script didn't seem to be installing it in an accessible way?)
• it tries to gracefully handle cluster shutdown in the event of error states, but still isn't going to be super robust to a VM pre-emption type kill, or a job deletion. Hopefully the low max_idle will act as a backstop there

This does need to be tested. Currently I've got a branch on the go here - I've built a new docker image containing this feature branch and will see how it behaves when replacing the metamist registration/dataproc interaction scripts.

--

Unrelated - it turns out the doctests in this repository were just cosmetic, pytest didn't look for them. The test/test_doctests.py file was doing nothing so that's gone. Instead the doctests are registered in pyproject.toml so pytest knows to look for them.

[Alexander-Stuckey]

Some things to think about, looks pretty good to start with

[Alexander-Stuckey]

I kinda didn't like this function, so I got Fable to pitch in. Needs a couple of new imports

from cloudpathlib import GSClient, GSPath


def upload_to_gcs(
    local_path: str,
    bucket: str,
    prefix: str = '',
    client: GSClient | None = None,
) -> str:
    bucket_name = bucket.removeprefix('gs://').rstrip('/')
    dest = GSPath(f'gs://{bucket_name}', client=client) / prefix.strip('/') / os.path.basename(local_path)
    dest.upload_from(local_path)
    return str(dest)

We can then do some finangling to allow upload from either file or memory (these belong in the class itself)

def _dest(self, prefix: str | None, name: str) -> GSPath:
    effective = prefix if prefix is not None else f'scripts/{self._name}'
    return GSPath(f'gs://{self._staging_bucket_name}', client=self._gs_client) / effective.strip('/') / name

def upload(self, local_path: str, prefix: str | None = None) -> str:
    dest = self._dest(prefix, os.path.basename(local_path))
    dest.upload_from(local_path)
    return str(dest)

def upload_bytes(self, data: bytes, name: str, prefix: str | None = None) -> str:
    dest = self._dest(prefix, name)
    dest.write_bytes(data)
    return str(dest)

[MattWellie]

I'm never sure how to feel about this. The google storage client feels like the blessed way of doing this, but it's pretty long-winded... Equally if we lean in on cloudpathlib (like we do elsewhere in cpg_utils/flow) this can be even simpler - e.g. using AnyPath to obfuscate between local and GCP paths. I think I like the idea of being able to upload from memory, but can't quite grok how that would look in practice from this suggestion, could you sketch it out a bit more?

[Alexander-Stuckey]

So cloudpathlib does exactly the same blob.upload_from_filename under the hood, so it's much of the same, but (subjectively) easier to read over?

Regarding the in-memory upload, thinking more about it, it seems that if something like a PED file is needed, that could just be generated inside the Dataproc cluster?

So perhaps we can just ditch the whole 'in memory' part

[MattWellie]

Cool, ok. Pushing on with this as the MVP which we can potentially extend later, I've adopted the GSPath/Client version of the method. Will try and get a test run going, see how it behaves

[Alexander-Stuckey]

Do you really want a bare Exception? Surely these is a tighter class that can be used

[MattWellie]

Suggested change

	Entries whose key does not start with a lowercase letter after sanitisation
	are dropped with a warning to stderr, matching the GCP label constraints.
	Entries whose sanitised key does not start with a lowercase letter causes an error.

[MattWellie]

Suggested change

	Intended for CLI arguments. Malformed entries missing an equals sign are
	skipped with a warning to stderr.
	Intended to format CLI-passed arguments. Malformed entries missing an equals sign cause an error.

[MattWellie]

Suggested change

	if any(value is None for value in [project, region, policy_ref]):
	if not all(project, region, policy_ref):

Empty strings here would also break, so run an any() truthiness test