Merge pull request #322 from sunya-ch/ec2

update instance list and pipeline for ec2-0.7.11

Author: sunya-ch

cmd/cmd_plot.py

@@ -155,7 +155,10 @@ def metadata_plot(args, energy_source, metadata_df, output_folder, name):
     plt.close()
 
 def power_curve_plot(args, data_path, energy_source, output_folder, name):
-    model_toppath = data_path
+    if 'MODEL_PATH' in os.environ:
+        model_toppath = os.environ['MODEL_PATH']
+    else:
+        model_toppath = data_path
     pipeline_name = args.pipeline_name
     pipeline_path = os.path.join(model_toppath, pipeline_name)
     node_collection = NodeTypeIndexCollection(pipeline_path)

cmd/cmd_util.py

@@ -28,7 +28,7 @@ def print_file_to_stdout(data_path, args):
 
 def extract_time(data_path, benchmark_filename):
     data = load_json(data_path, benchmark_filename)
-    if benchmark_filename != "customBenchmark":
+    if "metadata" in data:
         start_str = data["metadata"]["creationTimestamp"]
         start = datetime.datetime.strptime(start_str, '%Y-%m-%dT%H:%M:%SZ')
         end_str = data["status"]["results"][-1]["repetitions"][-1]["pushedTime"].split(".")[0]

cmd/main.py

@@ -394,6 +394,7 @@ def train(args):
     dyn_trainer_names = args.dyn_trainers.split(",")
 
     node_type=None
+    pipeline=None
     if args.id:
         machine_id = args.id
         pipeline = get_pipeline(data_path, pipeline_name, args.extractor, args.profile, args.target_hints, args.bg_hints, args.abs_pipeline_name, args.isolator, abs_trainer_names, dyn_trainer_names, energy_sources, valid_feature_groups)
@@ -749,10 +750,16 @@ def export(args):
         inputs = args.input.split(",")
 
     pipeline_name = args.pipeline_name
-    pipeline_path = get_pipeline_path(data_path, pipeline_name=pipeline_name)
+    if 'MODEL_PATH' not in os.environ:
+        os.environ['MODEL_PATH'] = data_path
+    pipeline_path = get_pipeline_path(os.environ['MODEL_PATH'], pipeline_name=pipeline_name)
 
     local_export_path = exporter.export(data_path, pipeline_path, output_path, publisher=args.publisher, collect_date=collect_date, inputs=inputs)
 
+    if local_export_path is None:
+        print("failed to export")
+        exit()
+
     args.input = local_export_path
     args.output = local_export_path
     args.energy_source = ",".join(PowerSourceMap.keys())

src/train/ec2_pipeline.py

@@ -12,7 +12,7 @@
 DATAPATH=/path/to/models python cmd/main.py plot --target-data estimate --input /path/to/data/i3.metal/kepler_query.json --pipeline-name ec2 --energy-source rapl-sysfs --model-name LogarithmicRegressionTrainer_4 --output-type AbsPower --output i3metal-ec2 --feature-group BPFOnly
 
 example of export command:
-DATAPATH=/path/to/models python cmd/main.py export --pipeline-name ec2 -o /path/to/kepler-model-db/models --publisher sunya-ch --zip=true --collect-date "July 2024"
+DATAPATH=/path/to/models python cmd/main.py export --pipeline-name ec2-0.7.11 -o /path/to/kepler-model-db/models --publisher sunya-ch --zip=true --collect-date "July 2024"
 """
 
 import os
@@ -40,10 +40,13 @@
 from saver import save_json
 from config import model_toppath
 
-node_profiles = ["m5zn.metal", "c5d.metal", "i3en.metal", "m7i.metal-24xl", "i3.metal"]
+data_path = os.path.join(model_toppath, "..", "data")
+
+node_profiles = ["m5.metal", "i3.metal", "c5.metal",  "r5.metal", "m5zn.metal", "m7i.metal-24xl"]
 node_image = "ami-0e4d0bb9670ea8db0"
 
 import boto3
+last_modified = None
 
 aws_access_key_id = os.environ["AWS_ACCESS_KEY_ID"]
 aws_secret_access_key = os.environ["AWS_SECRET_ACCESS_KEY"]
@@ -56,8 +59,9 @@
 def read_response_in_json(key):
     print(key)
     response = s3.get_object(Bucket=bucket_name, Key=key)
-    last_modified_str = response['LastModified']
-    print("{} last modified time: {}".format(key, last_modified_str))
+    global last_modified
+    last_modified = response['LastModified']
+    print("{} last modified time: {}".format(key, last_modified))
     return json.loads(response['Body'].read().decode('utf-8'))
 
 class Ec2PipelineRun():
@@ -77,15 +81,15 @@ def process(self):
             spec_json = read_response_in_json(spec_key)
 
             # save raw data from response in local path models/../data/<instance profile>
-            profile_datapath = os.path.join(model_toppath, "..", "data", profile)
+            profile_datapath = os.path.join(data_path, profile)
             os.makedirs(profile_datapath, exist_ok=True)
             save_json(path=profile_datapath, name="kepler_query.json", data=query_response)
-            save_json(path=profile_datapath, name=machine_spec_filename, data=spec_json)
+            machine_spec_path = os.path.join(profile_datapath, "machine_spec")
+            save_json(path=machine_spec_path, name=machine_spec_filename, data=spec_json)
 
             # process pipeline
-            spec = NodeTypeSpec(**spec_json['attrs'])
-            spec.attrs[NodeAttribute.MEMORY] = unknown
-            spec.attrs[NodeAttribute.FREQ] = unknown
+            spec = NodeTypeSpec()
+            spec.load(spec_json)
             node_type = self.pipeline.node_collection.index_train_machine(machine_id, spec)
             query_results = prom_responses_to_results(query_response)
             query_results[node_info_column] = node_type
@@ -101,8 +105,14 @@ def archive_pipeline(self):
         self.pipeline.archive_pipeline()
 
 if __name__ == "__main__":
-    pipeline_name = "ec2"
+    pipeline_name = "ec2-0.7.11"
     pipelinerun = Ec2PipelineRun(name=pipeline_name)
     pipelinerun.process()
     pipelinerun.save_metadata()
-    pipelinerun.archive_pipeline()
+    pipelinerun.archive_pipeline()
+    print("Collection time:", last_modified)
+    item = dict()
+    item["startTimeUTC"] = last_modified.strftime("%Y-%m-%dT%H:%M:%SZ")
+    item["endTimeUTC"] = last_modified.strftime("%Y-%m-%dT%H:%M:%SZ")
+    print(item)
+    save_json(path=data_path, name=pipeline_name, data=item)

src/train/profiler/node_type_index.py

@@ -2,7 +2,7 @@
 #p
 # Usage:
 #   index_collection = NodeTypeIndexCollection(pipeline_path)
-#   node_type = index_collection.index_train_machine(processor, cores, chips, cache_kb, memory_gb, cpu_freq_mhz, power_mgt, minmax_watt)
+#   node_type = index_collection.index_train_machine(machine_id, new_spec)
 #   index_collection.save()
 
 import sys
@@ -68,8 +68,8 @@ def generate_spec(data_path, machine_id):
         "processor": processor,
         "cores": cores,
         "chips": chips,
-        "memory_gb": memory_gb,
-        "cpu_freq_mhz": cpu_freq_mhz,
+        "memory": memory_gb,
+        "frequency": cpu_freq_mhz,
         "threads_per_core": threads_per_core
     }
     spec = NodeTypeSpec(**spec_values)
@@ -102,8 +102,8 @@ def __init__(self, **kwargs):
         self.attrs[NodeAttribute.PROCESSOR] = kwargs.get('processor', no_data)
         self.attrs[NodeAttribute.CORES] = kwargs.get('cores', no_data)
         self.attrs[NodeAttribute.CHIPS] = kwargs.get('chips', no_data)
-        self.attrs[NodeAttribute.MEMORY] = kwargs.get('memory_gb', no_data)
-        self.attrs[NodeAttribute.FREQ] = kwargs.get('cpu_freq_mhz', no_data)
+        self.attrs[NodeAttribute.MEMORY] = kwargs.get('memory', no_data)
+        self.attrs[NodeAttribute.FREQ] = kwargs.get('frequency', no_data)
         self.members = []
 
     def load(self, json_obj):

tests/pipeline_test.py

@@ -21,8 +21,8 @@
                     "processor": "test",
                     "cores": 1,
                     "chips": 1,
-                    "memory_gb": -1,
-                    "cpu_freq_mhz": -1
+                    "memory": -1,
+                    "frequency": -1
                 }
 spec = NodeTypeSpec(**spec_values)