def execute()

in gcpdiag/runbook/vertex/generalized_steps.py [0:0]
58 lines of code
6 McCabe index (conditional complexity)

  def execute(self):
    """Verify Jupyter is running on port 127.0.0.1:8080"""
    start_time = op.get(flags.START_TIME)
    end_time = op.get(flags.END_TIME)
    project_id: str = self.project_id or op.get(flags.PROJECT_ID)
    instance_name: str = self.instance_name or op.get(flags.INSTANCE_NAME)
    zone: str = self.zone or op.get(flags.ZONE)
    workbench_instance: notebooks.WorkbenchInstance = notebooks.get_workbench_instance(
        project_id=project_id, zone=zone, instance_name=instance_name)
    filter_str = r'''severity=INFO
                      AND
                      resource.type="gce_instance"
                      AND
                      log_name=~"projects\/{project_id}\/logs\/serialconsole.googleapis.com.*"
                      AND
                      labels."compute.googleapis.com/resource_name"="{instance_name}"
                      AND
                      textPayload=~"ServerApp.*Jupyter Server.*running at.*"'''.format(
        project_id=project_id, instance_name=instance_name)
    serial_log_entries_jupyter_running = logs.realtime_query(
        project_id=project_id,
        filter_str=filter_str,
        start_time=start_time,
        end_time=end_time)
    if serial_log_entries_jupyter_running:
      op.info(
          'Jupyter is running! Verifying if it\'s running on port 127.0.0.1:8080.'
      )
      filter_str = r'''severity=INFO
                    AND
                    resource.type="gce_instance"
                    AND
                    log_name=~"projects\/{project_id}\/logs\/serialconsole.googleapis.com.*"
                    AND
                    labels."compute.googleapis.com/resource_name"="{instance_name}"
                    AND NOT
                    textPayload=~"ServerApp.*localhost:8080\/lab"
                    AND
                    textPayload=~"ServerApp.*localhost:[0-9]{{4}}\/lab"'''.format(
          project_id=project_id, instance_name=instance_name)
      serial_log_entries_jupyter_port = logs.realtime_query(
          project_id=project_id,
          filter_str=filter_str,
          start_time=start_time,
          end_time=end_time)
      if serial_log_entries_jupyter_port:
        #User will need to fix their instance o create a new one
        op.add_failed(resource=workbench_instance,
                      reason=op.prep_msg(op.FAILURE_REASON_ALT1,
                                         instance_name=instance_name),
                      remediation=op.prep_msg(op.FAILURE_REMEDIATION))
      else:
        op.add_ok(resource=workbench_instance,
                  reason=op.prep_msg(op.SUCCESS_REASON,
                                     instance_name=instance_name))
    else:
      #User needs to make sure their instance is not stopped to get logs
      op.add_failed(resource=workbench_instance,
                    reason=op.prep_msg(op.FAILURE_REASON,
                                       instance_name=instance_name),
                    remediation=op.prep_msg(op.FAILURE_REMEDIATION))