In T309699#8903606, @Ottomata wrote:

@gmodena what should we do with the page_content_change event when we get badrevids? Right now, we discard it, so e.g. the page create event from page_change won't exist in the page_content_change stream.

I have a feeling we should just produce it as is into page_content_change with no content...or maybe some way of indicating that we could not get the content? Not sure though. cc also @Milimetric

Project documentation is available at
https://doc.wikimedia.org/data-engineering/eventutilities-python/eventutilities_python/

@tchin an alternative path for coverage reporting could be integrating with https://gitlab.wikimedia.org/repos/releng/docpub/-/blob/main/README.md and linking the published coverage report from Gitlab (we'd lose metric reporting in badge - but so be it).

Couple of questions re integrating this workflow in our pipeline. See the attached patch and this pipeline.

• As a test, I wanted to trigger .docpub:publish-docs (derived) manually from a working (unprotected) branch, but I'm not authorized to do so. Will this job automatically execute on a protected branch?
• We do releases by manually triggering a publish_gitlab_release on main. If I wanted to automatically trigger .docpub:publish-docs on release, can I just declare a need: publish_gitlab_release condition?

Think we can publish pydocs as part of this task!?

I marked the Google Doc as read-only and moved the draft to https://wikitech.wikimedia.org/wiki/MediaWiki_Event_Enrichment/SLO/Mediawiki_Page_Content_Change_Enrichment.

mw_page_content_change_enrich__dse-k8s-eqiad is not a valid s3 bucket, because the protocol does no allow _ in names: https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html.
As a test, I replaced _ with - and checkpoints are now stored.

This log entry is also relevant to the error above:

com.amazonaws.services.s3.model.AmazonS3Exception: The specified bucket is not valid. (Service: Amazon S3; Status Code: 400; Error Code: InvalidBucketName; Request ID: tx07b821592df14024bd0fb-00646dbe61; S3 Extended Request ID: null; Proxy: null), S3 Extended Request ID: null

There's a draft at https://docs.google.com/document/d/1U2bYVqmEsn7ryP0dtFUr-S5xPqF9_plLIFdzk883HBc/edit#. After a first round of feedback, I'll move it to wikitech.

Hopping on this thread to confirm that we are now able to store snapshots / savepoints in swift using the provided containers.

In T336656#8874300, @Ottomata wrote:

@gmodena I think(?) I've deployed in dse-k8s-eqiad staging. HA has been disabled, but swift checkpointing should be enabled? I'm not entirely sure how to check though.

As for other docs to move: probably all links at
https://www.mediawiki.org/wiki/Platform_Engineering_Team/Event_Platform_Value_Stream (expect maybe for the RFCs on Use Cases?).

There is some deployment documentation (dse-k8s) at https://www.mediawiki.org/wiki/Platform_Engineering_Team/Event_Platform_Value_Stream/Pyflink_Enrichment_Service_Deployment.

@Ottomata should we adopt this naming conventions also for DSE?

In T336656#8861728, @Ottomata wrote:

s3://<namespace>-<k8s-cluster-name-env>/<release>/flink/

Let's go with flink <job_name> instead of <namespace>. In https://phabricator.wikimedia.org/T330507#8860655 I just ran into an issue where the helmfile/namespace name is too long for our WMF prod k8s/helm conventions, and in that case,

In T331283#8858931, @Ottomata wrote:

As we have one zookeeper cluster per DC, I think it's not required to include the DC name in the Zookeeper path.

Hm, true, but we probably shouldn't try to un-suffix the k8s cluster names. It is redundant, but it is probably better if we just refer to clusters as their full names.

In T331283#8858396, @JMeybohm wrote:

In T322125#8857298, @Ottomata wrote:

@gmodena can we close this task?

Checkpointing to Swift (S3 protocol) has been enabled.

Flink docs recommend setting zookeeper https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#high-availability-zookeeper-path-root. I need some help to brainstorm / bikeshed a naming convention.

@elukey @JMeybohm since it seems we reached consensus, I'd like to enable Zookeeper HA in our app. Who's currently responsible for Zookeeper? Is there a request process I should follow, or can I just go ahead and configure the application helmfile?

In T330693#8846250, @Eevans wrote:

Ok, this is setup and has been tested. I created the two containers discussed as well (mediawiki-page-content-change-enrichment-{eqiad,codfw}).

Discussed with @Ottomata, and we opted for option 2: We log and raise a more functional error message.

We have a seemingly stable job running on YARN. Using an ad hoc process function that combines count and time triggers on a keyed data stream, seem to have reduced memory pressure on the beam workers.

In T330693#8701662, @gmodena wrote:

[...]

@MatthewVernon brought up persistent volume claim. I have no experience with it, but it sounds like a good fit; I'd like to explore that a bit.

I will investigate.

Next step will be measuring latency/throughput on YARN and possibly tune settings (batch size, thread pool size). If this works, we can look at integrating it into eventutilities-python.

Caveat: on DSE we run the application with the same memory settings, and noticed a slightly higher memory footprint (in the order of 20%). Might be due to different Java versions (java8 on YARN, java11 on k8s). We have tasks for load testing and tuning planned for upcoming sprints.

I have been able to validate that the idea can work. I was able to init a thread pool local to an operator and execute parallel requests on a batch of elements (PoC implementation here). No pickling issues, because what is put on the wire is the function output (instead of the closure). I think we could fallback to using a ProcessWindowFunction for compute + managing sideoutput, without too much impact on our API. I have a rough implementation outside of eventutilities-python, that I got to work locally (docker/minikube). Next step will be measuring latency/throughput on YARN and possibly tune settings (batch size, thread pool size). If this works, we can look at integrating it into eventutilities-python.

After a few long lasting runs on YARN and k8s all I can say is I see correlation (not necessarilly causation!) with OOMs and:

1. calls to dict_to_row conversion
2. python fn-bundle size (beam mico-batching)
3. messages throughput and kafka topic lag

To some degree I think that blocking I/O is contributing to memory pressure (saying this because of the effect of tuning microbatching via fn-bundle size)

I created https://phabricator.wikimedia.org/T332948 as follow up work for this spike.

I experimented with tunables recommended on Flink's Slack, and was able to run the application for over 6 hours. Memory kept growing till container OOM.
Some additional info that I was able to observe:

1. memory consumption grows over time, the number of objects tracked by Python's GC seem to stabilise after a few minutes of runtime, and not grow linearly with memory.
2. How quickly memory grows seems to be proportional to python.fn-execution.bundle.size

I enabled the python profiler (python.profile.enabled)
and was able to identify some hot paths that do correlate to memory
allocation (SerDes and dict <-> Row conversion).

Hi Eric,

In T330693#8698639, @MatthewVernon wrote:

This is a k8s application running on the WMF OpenStack, yes?

Might it be appropriate to use a persistent volume claim, backed by the OpenStack storage itself for this? At my last job that's the sort of solution we'd have looked at for this kind of workflow. That would give you small amounts of fast storage local to your compute environment.

In T330693#8696219, @Eevans wrote:

So: since we decided to use plain on datastream.map when error_destination = False (instead of our datastream.process w error handler), the job was dying when it encountered these errors. This makes sense I think. If stream manager error handling is disabled, then it is up to the user to catch exceptions in their own map function, right?