OTEL flow run context propagation with Labels

[jeanluciano]

This PR adds OTEL context propagation when __OTEL_TRACEPARENTS are found in a parent flow run. It also adds a way to update flow run labels with update_flow_run_labels. This is so that if no __OTEL_TRACEPARENT is found we can update the label with the newly created span's context.

Checklist

• This pull request references any related issue by including "closes <link to issue>"
  • If no issue exists and your change is not a small fix, please create an issue first.
• If this pull request adds new functionality, it includes unit tests that cover the changes
• If this pull request removes docs files, it includes redirect settings in mint.json.
• If this pull request adds functions or classes, it includes helpful docstrings.

[codspeed-hq]

CodSpeed Performance Report

Merging #16122 will not alter performance

_{Comparing jean/cloud-734-move-update_flow_run_labels-to-standard-client (653eb55) with main (8fd92fe)}

Summary

✅ 3 untouched benchmarks

[zzstoatzz]

a couple comments to start

[zzstoatzz]

looking pretty good! one more comment

[desertaxle]

Might be worthwhile to parameterize this with a sync and async flow now that the engine is split.

[jeanluciano]

I have it done in an upstream PR #16233

[desertaxle]

Might be worthwhile to parameterize this with a sync and async flow now that the engine is split.

[jeanluciano]

Same as above.

[desertaxle]

Two small requests on the orchestration client changes, but otherwise LGTM!

[desertaxle]

Let's not return the response from here since it looks like it isn't needed.

Suggested change

	async def update_flow_run_labels(
	self, flow_run_id: UUID, labels: KeyValueLabelsField
	) -> httpx.Response:
	"""
	Updates the labels of a flow run.
	"""
	response = await self._client.patch(
	f"/flow_runs/{flow_run_id}/labels", json=labels
	)
	response.raise_for_status()
	return response
	async def update_flow_run_labels(
	self, flow_run_id: UUID, labels: KeyValueLabelsField
	) -> None:
	"""
	Updates the labels of a flow run.
	"""
	response = await self._client.patch(
	f"/flow_runs/{flow_run_id}/labels", json=labels
	)
	response.raise_for_status()

[desertaxle]

Ditto

Suggested change

	def update_flow_run_labels(
	self, flow_run_id: UUID, labels: KeyValueLabelsField
	) -> httpx.Response:
	"""
	Updates the labels of a flow run.
	"""
	response = self._client.patch(
	f"/flow_runs/{flow_run_id}/labels",
	json=labels,
	)
	response.raise_for_status()
	return response
	def update_flow_run_labels(
	self, flow_run_id: UUID, labels: KeyValueLabelsField
	) -> None:
	"""
	Updates the labels of a flow run.
	"""
	response = self._client.patch(
	f"/flow_runs/{flow_run_id}/labels",
	json=labels,
	)
	response.raise_for_status()