example_tpv_config_locations_api.yml

global:
  default_inherits: default

tools:
  default:
    cores: 2
    mem: 8
    env: {}
    scheduling:
      require: []
      prefer:
          general
      accept:
      reject:
          offline
    rules: []
    rank: |
      import requests
      import json
      import pathlib
      from ruamel.yaml import YAML
      from galaxy import model
      from sqlalchemy import select, func
      from datetime import datetime, timedelta

      sa_session = app.model.context

      # NOTE: currently object store info is stored in a yaml
      objectstore_loc_path = "{{ tpv_config_dir }}/object_store_locations.yml"
      dataset_attributes = helpers.get_dataset_attributes(job.input_datasets)

      yaml=YAML(typ='safe')
      objectstore_file = pathlib.Path(objectstore_loc_path)
      objectstore_dict = yaml.load(objectstore_file)

      # URL of local influxDB
      influx_url = "http://127.0.0.1:8086/query"
      # URL of the FastAPI application
      api_url = "http://localhost:8000/process_destinations"
      queries = [
          "SELECT * FROM \"destination-queue-run-time\"  WHERE time > now() - 30m",
          "SELECT * FROM \"cluster.queue\" WHERE host = 'vgcn-pulsar-central-manager.usegalaxy.be' LIMIT 6;",
          "SELECT * FROM \"cluster.alloc\" WHERE host = 'vgcn-pulsar-central-manager.usegalaxy.be' LIMIT 10;",
      ]

      params = {
         "db": "telegraf",
         "q": ";".join(queries)
      }
      response = requests.get(influx_url, params=params)
      results = response.json()

      columns = results['results'][0]['series'][0]['columns']
      stat_columns = ['destination_id', 'tool_id', 'time', 'count', 'median_queue', 'median_run']
      stat_indices = [columns.index(c) for c in stat_columns]
      #stat_indices = [i for i,l in enumerate(columns) if l in stat_columns]
      series = results['results'][0]['series'][0]['values']

      #condor_status = [zip(results['results'][1]['series'][0]['columns'], row) for row in results['results'][1]['series'][0]['values'])]

      # Format the destination data and identify the total number of queued jobs for each destination
      candidate_destinations_list = []
      for dest in candidate_destinations:
          dest_dict = dest.to_dict()
          dest_dict["queued_job_count"] = app.model.context.query(model.Job).filter(model.Job.state == "queued", model.Job.destination_id == dest.dest_name).count()

          for row in series:
              stats = [row[i] for i in stat_indices]
              if stats[0] == dest.dest_name and stats[1] == job.tool_id:
                  dest_dict.update(zip(stat_columns,stats))
                  dest_dict["galaxy_db_query_time"] = dest_dict.pop("time")
                  dest_dict["job_count_in_time_window"] = dest_dict.pop("count")
                  dest_dict["median_queue_time"] = dest_dict.pop("median_queue")
                  dest_dict["median_run_time"] = dest_dict.pop("median_run")
                  break
          
          for row in results['results'][1]['series'][0]['values']:
              dest_dict[row[-1]+"_count"] = row[1]

          for row in results['results'][2]['series'][0]['values']:
              dest_dict["cpu_usage_perc"] = row[2]
              dest_dict["mem_usage_perc"] = row[-1]

          candidate_destinations_list.append(dest_dict)

      input_data = {
        "destinations": candidate_destinations_list,
        "objectstores": objectstore_dict,
        "dataset_attributes": dataset_attributes
      }

      # Send a POST request to the API endpoint
      response = requests.post(api_url, json=input_data)

      # Check if the request was successful (status code 200)
      if response.status_code == 200:
          result = response.json()
      else:
          print(f"Request failed with status code {response.status_code}: {response.text}")

      final_destinations = sorted(candidate_destinations, key=lambda obj: result["sorted_destinations"].index(str(obj.id)))

      final_destinations

  trinity:
    cores: 2
    mem: cores * 4
    env: {}
    scheduling:
      require: []
      prefer:
        - pulsar
      accept:
      reject:
        - offline

roles:
  ga_admins:
    scheduling:
      require:
        []


destinations:
  pulsar_italy:
    runner: general_pulsar_1
    max_accepted_cores: 8
    max_accepted_mem: 32
    scheduling:
      accept:
        - general
      require:
        - pulsar
    context:
      latitude: 50.0689816
      longitude: 19.9070188
  slurm_poland:
    runner: slurm
    max_accepted_cores: 16
    max_accepted_mem: 64
    scheduling:
      accept:
        - slurm
    context:
      latitude: 51.9189046
      longitude: 19.1343786
  condor_belgium:
    runner: condor
    max_accepted_cores: 16
    max_accepted_mem: 64
    scheduling:
      accept:
        - condor
    context:
      latitude: 50.5010789
      longitude: 4.4764595
  slurm_germany:
    runner: slurm
    max_accepted_cores: 16
    max_accepted_mem: 64
    scheduling:
      accept:
        - slurm
    context:
      latitude: 51.1642292
      longitude: 10.4541192
  condor_france:
    runner: condor
    max_accepted_cores: 16
    max_accepted_mem: 64
    scheduling:
      accept:
        - condor
    context:
      latitude: 46.71109
      longitude: 1.7191036
  pulsar_australia:
    runner: general_pulsar_1
    max_accepted_cores: 8
    max_accepted_mem: 32
    scheduling:
      accept:
        - general
      require:
        - pulsar
    context:
      latitude: -26.4390917
      longitude: 133.281323