How to access the run event logs of all code locations under dagster helm deployment

[Lukashab]

Hello,
I have a dagster deployed with dagster helm chart where I have several code locations using postgres as an event storage.
Since there is a lot of jobs running every minute, the postgres filled up quite quickly, so I came up with a solution how to rotate the event log data so the storage will not crash on "disk empty" error.

I created a new code location with "rotate" job that runs every hour and cleans up records that are older than x days.
My code is as follows:

@op
def delete_old_event_logs(metrics_client: MetricsClient) -> None:
    """
    Deletes event logs from storage that are older than EVENT_LOG_ROTATION_DAYS
    """
    instance = DagsterInstance.get()
    date_from = utcnow() - timedelta(days=settings.EVENT_LOG_ROTATION_DAYS)
    batch_size = 100

    runs_deleted = 0
    deleted_metric = EventLogDeletedGauge(
        prometheus=metrics_client.prometheus
    ).metric.labels(environment=settings.ENV_FOR_DYNACONF)

    while True:
        old_run_records = instance.get_run_records(
            filters=RunsFilter(created_before=date_from),
            ascending=True,
            limit=batch_size,
        )

        if len(old_run_records) == 0:
            break

        for record in old_run_records:
            instance.delete_run(record.dagster_run.run_id)

        deleted_metric.inc(len(old_run_records))

    logger.info(f"Deleted {runs_deleted} runs")

The problem is that, for some reason, it never really deletes the records for other code locations. It seems that, maybe, the dagster instance in this context is only the instance related to this one code location that the job is running in? It is really hard to debug, so my question is. Is this a good approach and what could be the problem here?
I will be glad for any advice.

[prha]

I think this looks like it could work, but some things you might want to try:

• Ensure that you're accessing the correct instance. I think what you have should work, but would confirm that the instance configuration off of DagsterInstance.get matches the instance provided off of the OpExecutionContext (which is provided to the op if you add a context parameter. See https://docs.dagster.io/_apidocs/execution#dagster.OpExecutionContext.instance)
• Ensure that you can fetch the run records that you expect. You should be able to open a Python interpreter in the environment, call DagsterInstance.get() and make the get_run_records call with the filter you want. This could help you debug in a local setting that you're getting the records you want.

[Lukashab]

@prha thank you. You were right. I changed the code so the instance is obtained from the op context variable like context.instance and it works as expected now.
Thank you for all the insights and advice