anonymizer stage flow (#2202)

Summary:
Pull Request resolved: #2202

## What

- Create a stageflow for running the anonymizer game

## Why

- As part of the Q1 PD "orchestration of workflows" prototype, I am creating some skeletons inside of PCS such that the PCF team would be able to integrate with PCS with minimal assistance from the PCA and/or PSI teams (assuming MPC infra is used in the future)
- PD "orchestration of workflows" doc: https://docs.google.com/document/d/1Nnn3rrXB1PokGl8a_fhB2JulXbLyHxpZe3-fPGqJO30/edit

Reviewed By: adshastri

Differential Revision:
D44149317

Privacy Context Container: L416713

fbshipit-source-id: 285be083d6e1555ac5ad47fdc70c0139a5f6a32f

Author: jrodal98

fbpcs/bolt/constants.py

@@ -13,6 +13,9 @@
 from fbpcs.private_computation.entity.private_computation_status import (
     PrivateComputationInstanceStatus,
 )
+from fbpcs.private_computation.stage_flows.private_computation_anonymizer_stage_flow import (
+    PrivateComputationAnonymizerStageFlow,
+)
 from fbpcs.private_computation.stage_flows.private_computation_base_stage_flow import (
     PrivateComputationBaseStageFlow,
 )
@@ -33,6 +36,7 @@
     PrivateComputationGameType.ATTRIBUTION: PrivateComputationPCF2StageFlow,
     PrivateComputationGameType.LIFT: PrivateComputationStageFlow,
     PrivateComputationGameType.PRIVATE_ID_DFCA: PrivateComputationPrivateIdDfcaStageFlow,
+    PrivateComputationGameType.ANONYMIZER: PrivateComputationAnonymizerStageFlow,
 }
 DEFAULT_MAX_PARALLEL_RUNS = 10
 DEFAULT_NUM_TRIES = 4

fbpcs/private_computation/stage_flows/__init__.py

@@ -28,6 +28,7 @@
     "private_computation_private_id_dfca_stage_flow",
     "private_computation_pid_continuous_measurement_stage_flow",
     "private_computation_pa_for_pd_stage_flow",
+    "private_computation_anonymizer_stage_flow",
 ]
 
 from . import *  # noqa: ignore=F403

fbpcs/private_computation/stage_flows/private_computation_anonymizer_stage_flow.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+from fbpcs.private_computation.entity.private_computation_status import (
+    PrivateComputationInstanceStatus,
+)
+from fbpcs.private_computation.service.anonymizer_stage_service import (
+    AnonymizerStageService,
+)
+from fbpcs.private_computation.service.private_computation_stage_service import (
+    PrivateComputationStageService,
+    PrivateComputationStageServiceArgs,
+)
+from fbpcs.private_computation.stage_flows.private_computation_base_stage_flow import (
+    PrivateComputationBaseStageFlow,
+    PrivateComputationStageFlowData,
+)
+
+
+class PrivateComputationAnonymizerStageFlow(PrivateComputationBaseStageFlow):
+    """
+    This enum lists all of the supported stage types and maps to their possible statuses.
+    It also provides methods to get information about the next or previous stage.
+
+    This should only be used to run the anonymizer with routine advertisers
+    """
+
+    # Specifies the order of the stages. Don't change this unless you know what you are doing.
+    # pyre-fixme[15]: `_order_` overrides attribute defined in `Enum` inconsistently.
+    _order_ = "CREATED ANONYMIZER"
+    # Regarding typing fixme above, Pyre appears to be wrong on this one. _order_ only appears in the EnumMeta metaclass __new__ method
+    # and is not actually added as a variable on the enum class. I think this is why pyre gets confused.
+
+    CREATED = PrivateComputationStageFlowData(
+        initialized_status=PrivateComputationInstanceStatus.CREATION_INITIALIZED,
+        started_status=PrivateComputationInstanceStatus.CREATION_STARTED,
+        completed_status=PrivateComputationInstanceStatus.CREATED,
+        failed_status=PrivateComputationInstanceStatus.CREATION_FAILED,
+        is_joint_stage=False,
+    )
+    ANONYMIZER = PrivateComputationStageFlowData(
+        initialized_status=PrivateComputationInstanceStatus.ANONYMIZER_INITIALIZED,
+        started_status=PrivateComputationInstanceStatus.ANONYMIZER_STARTED,
+        completed_status=PrivateComputationInstanceStatus.ANONYMIZER_COMPLETED,
+        failed_status=PrivateComputationInstanceStatus.ANONYMIZER_FAILED,
+        is_joint_stage=True,
+    )
+
+    def get_stage_service(
+        self, args: PrivateComputationStageServiceArgs
+    ) -> PrivateComputationStageService:
+        """
+        Maps PrivateComputationStageFlow instances to StageService instances
+
+        Arguments:
+            args: Common arguments initialized in PrivateComputationService that are consumed by stage services
+
+        Returns:
+            An instantiated StageService object corresponding to the StageFlow enum member caller.
+
+        Raises:
+            NotImplementedError: The subclass doesn't implement a stage service for a given StageFlow enum member
+        """
+        if self is self.ANONYMIZER:
+            return AnonymizerStageService(
+                args.onedocker_svc,
+                args.onedocker_binary_config_map,
+            )
+        else:
+            return self.get_default_stage_service(args)