Merge pull request #23 from sunya-ch/v1.0.1

add megatron parser

controllers/collector.go

@@ -185,7 +185,7 @@ func (c *ResultCollector) Collect(ch chan<- prometheus.Metric) {
  values := make(map[string]interface{})
  err := json.Unmarshal([]byte(item.Result), &values)
  if err != nil {
- c.Log.Info(fmt.Sprintf("Cannot parse values of %s from respone: %s", benchmarkName, item.Result))
+ c.Log.Info(fmt.Sprintf("Cannot parse values of %s from respone: %s: %v", benchmarkName, item.Result, err))
  continue
  }
  jobName := item.JobName

cpe-parser/api.go

@@ -35,6 +35,7 @@ var stressParser parser.Parser = parser.NewStressParser()
 var mlperfParser parser.Parser = parser.NewMlPerfParser()
 var fmworkParser parser.Parser = parser.NewFMWorkParser()
 var fmtrainParser parser.Parser = parser.NewFMTrainParser()
+var megatronParser parser.Parser = parser.NewMegatronParser()
 
 var parserMap map[string]parser.Parser = map[string]parser.Parser{
  "codait": codaitParser,
@@ -53,6 +54,7 @@ var parserMap map[string]parser.Parser = map[string]parser.Parser{
  "mlperf": mlperfParser,
  "fmwork": fmworkParser,
  "fmtrain": fmtrainParser,
+ "megatron": megatronParser,
 }
 
 /////////////////////////////////////////////
@@ -193,7 +195,8 @@ func ReqPushLog(w http.ResponseWriter, r *http.Request) {
  }
  }
  status = "OK"
- msg = fmt.Sprintf("%v", logSpec.ConstLabels)
+ dataBytes, _ := json.Marshal(values)
+ msg = string(dataBytes)
  pkey = ppkey
  pval = ppval
  }

cpe-parser/parser/megatron.go

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2022- IBM Inc. All rights reserved
+ * SPDX-License-Identifier: Apache2.0
+ */
+
+package parser
+
+import (
+ "bufio"
+ "bytes"
+ "io"
+ "strings"
+)
+
+type MegatronParser struct {
+ *BaseParser
+}
+
+/*
+2023-10-26 21:31:16,318 [Rank 15]: iteration 20/ 20 | consumed samples: 20 | elapsed time per iteration (ms): 3874.0 | learning rate: 6.000E-06 | global batch size: 16 | lm loss: 8.777018E+00 | loss scale: 1.0 | grad norm: 12.959 | number of skipped iterations: 0 | number of nan iterations: 0 | TFLOPs: 320.33 | tokens-per-second-per-gpu: 2114.61 |
+*/
+
+const (
+ MEGATRON_PERFORMANCE_KEY = "tokens-per-second-per-gpu"
+)
+
+func NewMegatronParser() *MegatronParser {
+ megatronParser := &MegatronParser{}
+ abs := &BaseParser{
+ Parser: megatronParser,
+ }
+ megatronParser.BaseParser = abs
+ return megatronParser
+}
+
+func (p *MegatronParser) ParseValue(body []byte) (map[string]interface{}, error) {
+ values := make(map[string]interface{})
+ valuesWithLabels := make(map[string][]ValueWithLabels)
+
+ bytesReader := bytes.NewReader(body)
+ bufReader := bufio.NewReader(bytesReader)
+ for {
+ line, _, err := bufReader.ReadLine()
+ linestr := string(line)
+ if err == io.EOF {
+ break
+ } else if err != nil {
+ return nil, err
+ }
+ if strings.Contains(linestr, MEGATRON_PERFORMANCE_KEY) {
+ labels := make(map[string]string)
+ splited := strings.Split(linestr, "|")
+ for _, col := range splited {
+ if strings.Contains(col, "Rank") && strings.Contains(col, "iteration") {
+ iterationFields := strings.Fields(col)
+ for _, field := range iterationFields {
+ if strings.Contains(field, "/") {
+ iter := strings.Split(field, "/")[0]
+ labels["iteration"] = iter
+ }
+ if strings.Contains(field, "]") {
+ rank := strings.Split(field, "]")[0]
+ labels["rank"] = rank
+ }
+ }
+ } else {
+ key, value, err := splitValue(col, ":")
+ if err == nil {
+ // trim key
+ keyFields := strings.Fields(key)
+ key = strings.Join(keyFields, " ")
+ newValue := ValueWithLabels{
+ Labels: labels,
+ Value: value,
+ }
+ if valueWithLabelsArr, ok := valuesWithLabels[key]; ok {
+ valuesWithLabels[key] = append(valueWithLabelsArr, newValue)
+ } else {
+ valuesWithLabels[key] = []ValueWithLabels{newValue}
+ }
+ }
+ }
+ }
+
+ }
+ }
+ for key, valueWithLabelsArr := range valuesWithLabels {
+ values[key] = valueWithLabelsArr
+ }
+ return values, nil
+}
+
+func (p *MegatronParser) GetPerformanceValue(values map[string]interface{}) (string, float64) {
+ if valuesWithLabelsInterface, ok := values[MEGATRON_PERFORMANCE_KEY]; ok {
+ valuesWithLabels := valuesWithLabelsInterface.([]ValueWithLabels)
+ if len(valuesWithLabels) == 0 {
+ return "NoValue", -1
+ }
+ avgValue := float64(0)
+ for _, valueWithLabel := range valuesWithLabels {
+ avgValue += valueWithLabel.Value
+ }
+ if len(valuesWithLabels) > 0 {
+ avgValue /= float64(len(valuesWithLabels))
+ }
+ return MEGATRON_PERFORMANCE_KEY, avgValue
+ }
+ return "NoKey", -1
+}

cpe-parser/parser/megatron_test.go

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2022- IBM Inc. All rights reserved
+ * SPDX-License-Identifier: Apache2.0
+ */
+
+// Run go test -v parser/megatron_test.go
+
+package parser
+
+import (
+ "fmt"
+ "io/ioutil"
+ "math"
+ "testing"
+
+ "github.com/IBM/cpe-operator/cpe-parser/parser"
+ "github.com/stretchr/testify/assert"
+)
+
+// update log key here
+const (
+ LOG_KEY string = "megatron"
+)
+
+func getFileName() string {
+ return fmt.Sprintf("sample/%s_pod_log.log", LOG_KEY)
+}
+
+var generalParser parser.Parser
+
+// update parser init function
+var testParser = parser.NewMegatronParser()
+
+func TestParseValue(t *testing.T) {
+ fileName := getFileName()
+ bytes, err := ioutil.ReadFile(fileName)
+ generalParser = testParser
+ assert.Nil(t, err)
+ values, err := generalParser.ParseValue(bytes)
+ assert.Nil(t, err)
+ // update assert value length
+ assert.Equal(t, len(values), 11)
+}
+
+func TestGetPerformanceValue(t *testing.T) {
+ fileName := getFileName()
+ bytes, err := ioutil.ReadFile(fileName)
+ generalParser = testParser
+ assert.Nil(t, err)
+ values, err := generalParser.ParseValue(bytes)
+ key, pvalue := testParser.GetPerformanceValue(values)
+ fmt.Printf("PKey: %s, Pvalue: %.2f\n", key, pvalue)
+ // update assert performance value
+ assert.Equal(t, math.Floor(pvalue), math.Floor((2206.58+2114.61)/2))
+}