Merge pull request #100 from paritytech/pib-onboard-inactive-para-task

Add task to onboard all inactive parachains

app/lib/cron_tasks.py

@@ -6,7 +6,8 @@
 
 from app.lib.kubernetes_client import list_stateful_sets, list_validator_stateful_sets
 from app.config.network_configuration import get_network, network_tasks_cron_schedule
-from app.lib.network_utils import rotate_nodes_session_keys, register_statefulset_validators
+from app.lib.network_utils import rotate_nodes_session_keys, register_statefulset_validators, list_parachains, \
+    onboard_parachain_by_id
 from app.lib.substrate import get_relay_chain_client
 
 log = logging.getLogger(__name__)
@@ -30,6 +31,10 @@ async def load_cron_tasks():
         scheduler.add_job(register_inactive_validators,
                           name='register_inactive_validators',
                           trigger=tasks_cron_trigger)
+        onboard_inactive_parachains = onboard_network_inactive_parachains()
+        scheduler.add_job(onboard_inactive_parachains,
+                          name='onboard_inactive_parachains',
+                          trigger=tasks_cron_trigger)
 
 
 async def exec_cron_task(job_id):
@@ -90,3 +95,17 @@ async def register_inactive_validators():
             print(stateful_set)
         log.info('Finished registering inactive validators')
     return register_inactive_validators
+
+
+def onboard_network_inactive_parachains():
+    async def onboard_inactive_parachains():
+        log.info(f'Onboarding inactive parachains for network={network}')
+        parachains = list_parachains()
+        for para_id, para_info in parachains.items():
+            # Onboard parachain if not currently active
+            if not para_info.get('lifecycle', '') in ['Parachain', 'Onboarding']:
+                await onboard_parachain_by_id(para_id, True)
+            else:
+                log.info(F'Parachain #{para_id} already onboarded')
+        log.info('Finished onboarding inactive parachains')
+    return onboard_inactive_parachains

app/lib/kubernetes_client.py

@@ -120,7 +120,7 @@ def list_validator_stateful_sets(role='authority'):
     return list(map(lambda sts: sts['metadata']['name'], validator_stateful_sets))
 
 
-def list_parachain_collator_stateful_sets(para_id):
+def list_parachain_collator_stateful_sets(para_id: str):
     stateful_sets = kubernetes_client.CustomObjectsApi().list_namespaced_custom_object(group="apps", version="v1", plural="statefulsets", namespace=namespace)
 
     collator_stateful_sets = list(
@@ -147,7 +147,7 @@ def list_validator_pods(stateful_set_name):
     return validator_pods
 
 
-def list_collator_pods(para_id=None, stateful_set_name=None):
+def list_collator_pods(para_id: str = None, stateful_set_name: str = None):
     collator_pods = list_substrate_node_pods('collator')
     if stateful_set_name:
         collator_pods = list(filter(lambda pod: pod.metadata.owner_references[0].name == stateful_set_name, collator_pods))

app/lib/network_utils.py

@@ -451,9 +451,9 @@ def list_parachains():
 
     # retrieve the list of parachains for which we have collators in the cluster
     collator_pods = list_collator_pods()
-    # If the paraId is not set on pod labels, we fallback to 0
+    # If the paraId is not set on pod labels, we fall back to 0
     cluster_parachain_tuple_set = set(
-        map(lambda pod: (int(pod.metadata.labels.get('paraId', 0)), pod.metadata.labels.get('chain')), collator_pods))
+        map(lambda pod: (pod.metadata.labels.get('paraId', '0'), pod.metadata.labels.get('chain')), collator_pods))
 
     parachains = {}
     for cluster_parachain_tuple in cluster_parachain_tuple_set:
@@ -483,31 +483,33 @@ def list_parachains():
     return parachains
 
 
-async def onboard_parachain_by_id(para_id, force_queue_action):
+async def onboard_parachain_by_id(para_id: str, force_queue_action: bool):
     log.info(f'starting to onboard parachain #{para_id}')
     relay_chain_client = get_relay_chain_client()
     sudo_seed = network_sudo_seed()
     parachain_pods = list_collator_pods(para_id)
-    para_node_client = get_node_client(parachain_pods[0].metadata.name)
-    node_para_id = get_parachain_id(parachain_pods[0])
-    if node_para_id == para_id:
-        state = get_parachain_head(para_node_client)
-        wasm = get_chain_wasm(para_node_client)
-
-        if state and wasm:
-            permanent_slot_lease_period_length = get_permanent_slot_lease_period_length(relay_chain_client)
-            log.info('Scheduling parachain #{}, state:{}, wasm: {}...{}, lease: {}'.format(
-                para_id, state, wasm[0:64], wasm[-64:], permanent_slot_lease_period_length))
-            initialize_parachain(relay_chain_client, sudo_seed, para_id, state, wasm, permanent_slot_lease_period_length, force_queue_action)
+    if parachain_pods:
+        para_node_client = get_node_client(parachain_pods[0].metadata.name)
+        node_para_id = get_parachain_id(parachain_pods[0])
+        if node_para_id == para_id:
+            state = get_parachain_head(para_node_client)
+            wasm = get_chain_wasm(para_node_client)
+            if state and wasm:
+                permanent_slot_lease_period_length = get_permanent_slot_lease_period_length(relay_chain_client)
+                log.info('Scheduling parachain #{}, state:{}, wasm: {}...{}, lease: {}'.format(
+                    para_id, state, wasm[0:64], wasm[-64:], permanent_slot_lease_period_length))
+                initialize_parachain(relay_chain_client, sudo_seed, para_id, state, wasm, permanent_slot_lease_period_length, force_queue_action)
+            else:
+                log.error(
+                    'Error: Not enough parameters to Scheduling parachain para_id: {}, state:{}, wasm: {}...{}'.format(
+                        para_id, state, wasm[0:64], wasm[-64:-1]))
         else:
-            log.error(
-                'Error: Not enough parameters to Scheduling parachain para_id: {}, state:{}, wasm: {}...{}'.format(
-                    para_id, state, wasm[0:64], wasm[-64:-1]))
+            log.error('Node para_id: {} doesn\'t match the requested offboard para_id {}'.format(node_para_id, para_id))
     else:
-        log.error('Node para_id: {} doesn\'t match the requested offboard para_id {}'.format(node_para_id, para_id))
+        log.error(f"Couldn't find parachain pod for para_id={para_id}")
 
 
-async def offboard_parachain_by_id(para_id, force_queue_action):
+async def offboard_parachain_by_id(para_id: str, force_queue_action: bool):
     log.info(f'starting to offboard parachain #{para_id}')
     substrate_client = get_relay_chain_client()
     sudo_seed = network_sudo_seed()
@@ -519,7 +521,7 @@ async def offboard_parachain_by_id(para_id, force_queue_action):
 
 
 # Collators
-def list_parachain_collators(para_id, stateful_set_name=''):
+def list_parachain_collators(para_id: str, stateful_set_name: str = ''):
     collator_pods = list_collator_pods(para_id, stateful_set_name)
     # Read the first collator pod chain metadata for this para_id to retrieve the chain name
     if collator_pods:
@@ -628,7 +630,7 @@ async def deregister_collator_nodes(chain, nodes, ss58_format):
     return accounts_to_deregister
 
 
-async def register_statefulset_collators(para_id, stateful_set_name):
+async def register_statefulset_collators(para_id: str, stateful_set_name: str):
     log.info('starting to register collators in statefulset {}'.format(stateful_set_name))
     collators_pods = list_collator_pods(para_id, stateful_set_name)
     chain = collators_pods[0].metadata.labels['chain']
@@ -637,7 +639,7 @@ async def register_statefulset_collators(para_id, stateful_set_name):
     await register_collator_nodes(chain, collator_node_names, ss58_format)
 
 
-async def deregister_statefulset_collators(para_id, stateful_set_name):
+async def deregister_statefulset_collators(para_id: str, stateful_set_name: str):
     log.info('starting to deregister collators in statefulset {}'.format(stateful_set_name))
     collators_pods = list_collator_pods(para_id, stateful_set_name)
     chain = collators_pods[0].metadata.labels['chain']

skaffold.yaml

@@ -24,4 +24,9 @@ portForward:
   resourceName: testnet-manager
   namespace: rococo
   port: 80
-  localPort: 8080
+  localPort: 8080
+- resourceType: service
+  resourceName: testnet-manager-task-scheduler
+  namespace: rococo
+  port: 80
+  localPort: 8081