[PH] Performance Harness csv improvements

tests/performance_tests/log_reader.py

@@ -29,6 +29,7 @@ class ArtifactPaths:
     trxGenLogDirPath: Path = Path("")
     blockTrxDataPath: Path = Path("")
     blockDataPath: Path = Path("")
+    csvDataPath: Path = Path("")
 
 @dataclass
 class TpsTestConfig:
@@ -62,6 +63,8 @@ class trxData():
     blockNum: int = 0
     cpuUsageUs: int = 0
     netUsageUs: int = 0
+    blockTime: datetime = None
+    latency: float = 0
     _sentTimestamp: str = ""
     _calcdTimeEpoch: float = 0
 
@@ -198,6 +201,7 @@ def selectedOpen(path):
     return gzip.open if path.suffix == '.gz' else open
 
 def scrapeLog(data: chainData, path):
+    #node_00/stderr.txt
     selectedopen = selectedOpen(path)
     with selectedopen(path, 'rt') as f:
         line = f.read()
@@ -228,16 +232,19 @@ def scrapeLog(data: chainData, path):
             data.forkedBlocks.append(int(fork[1]) - int(fork[3]) + 1)
 
 def scrapeTrxGenLog(trxSent, path):
+    #trxGenLogs/trx_data_output_*.txt
     selectedopen = selectedOpen(path)
     with selectedopen(path, 'rt') as f:
         trxSent.update(dict([(x[0], x[1]) for x in (line.rstrip('\n').split(',') for line in f)]))
 
 def scrapeBlockTrxDataLog(trxDict, path):
+    #blockTrxData.txt
     selectedopen = selectedOpen(path)
     with selectedopen(path, 'rt') as f:
-        trxDict.update(dict([(x[0], trxData(x[1], x[2], x[3])) for x in (line.rstrip('\n').split(',') for line in f)]))
+        trxDict.update(dict([(x[0], trxData(blockNum=x[1], blockTime=x[2], cpuUsageUs=x[3], netUsageUs=x[4])) for x in (line.rstrip('\n').split(',') for line in f)]))
 
 def scrapeBlockDataLog(blockDict, path):
+    #blockData.txt
     selectedopen = selectedOpen(path)
     with selectedopen(path, 'rt') as f:
         blockDict.update(dict([(x[0], blkData(x[1], x[2], x[3], x[4])) for x in (line.rstrip('\n').split(',') for line in f)]))
@@ -258,6 +265,17 @@ def populateTrxSentTimestamp(trxSent: dict, trxDict: dict, notFound):
         else:
             notFound.append(sentTrxId)
 
+def populateTrxLatencies(blockDict: dict, trxDict: dict):
+    for trxId, data in trxDict.items():
+        if data.calcdTimeEpoch != 0:
+            trxDict[trxId].latency = blockDict[data.blockNum].calcdTimeEpoch - data.calcdTimeEpoch
+
+def writeTransactionCsv(trxDict: dict, path):
+    with open(path, 'wt') as csvFile:
+        csvFile.write("TransactionId,BlockNumber,BlockTime,CpuUsageUs,NetUsageUs,Latency,SentTimestamp,CalcdTimeEpoch\n")
+        for trxId, data in trxDict.items():
+            csvFile.write(f"{trxId},{data.blockNum},{data.blockTime},{data.cpuUsageUs},{data.netUsageUs},{data.latency},{data._sentTimestamp},{data._calcdTimeEpoch}\n")
+
 def getProductionWindows(prodDict: dict, blockDict: dict, data: chainData):
     prod = ""
     count = 0
@@ -408,7 +426,7 @@ def calcTrxLatencyCpuNetStats(trxDict : dict, blockDict: dict):
     Returns:
     transaction latency stats as a basicStats object
     """
-    trxLatencyCpuNetList = [((blockDict[data.blockNum].calcdTimeEpoch - data.calcdTimeEpoch), data.cpuUsageUs, data.netUsageUs) for trxId, data in trxDict.items() if data.calcdTimeEpoch != 0]
+    trxLatencyCpuNetList = [(data.latency, data.cpuUsageUs, data.netUsageUs) for trxId, data in trxDict.items() if data.calcdTimeEpoch != 0]
 
     npLatencyCpuNetList = np.array(trxLatencyCpuNetList, dtype=np.float)
 
@@ -484,6 +502,8 @@ def calcAndReport(data: chainData, tpsTestConfig: TpsTestConfig, artifacts: Arti
         if argsDict.get("printMissingTransactions"):
             print(notFound)
 
+    populateTrxLatencies(blockDict, trxDict)
+    writeTransactionCsv(trxDict, artifacts.csvDataPath)
     guide = calcChainGuide(data, tpsTestConfig.numBlocksToPrune)
     trxLatencyStats, trxCpuStats, trxNetStats = calcTrxLatencyCpuNetStats(trxDict, blockDict)
     tpsStats = scoreTransfersPerSecond(data, guide)

tests/performance_tests/performance_test_basic.py

@@ -92,6 +92,12 @@ class SpecifiedContract:
         specificExtraNodeosArgs: dict = field(default_factory=dict)
         _totalNodes: int = 2
 
+        def log_transactions(self, trxDataFile, block):
+            for trx in block['payload']['transactions']:
+                for actions in trx['actions']:
+                    if actions['account'] != 'eosio' and actions['action'] != 'onblock':
+                        trxDataFile.write(f"{trx['id']},{trx['block_num']},{trx['block_time']},{trx['cpu_usage_us']},{trx['net_usage_words']},{trx['actions']}\n")
+
         def __post_init__(self):
             self._totalNodes = self.pnodes + 1 if self.totalNodes <= self.pnodes else self.totalNodes
             if not self.prodsEnableTraceApi:
@@ -105,7 +111,7 @@ def __post_init__(self):
                 self.specificExtraNodeosArgs.update({f"{node}" : '--plugin eosio::history_api_plugin --filter-on "*"' for node in range(self.pnodes, self._totalNodes)})
             else:
                 self.fetchBlock = lambda node, blockNum: node.processUrllibRequest("trace_api", "get_block", {"block_num":blockNum}, silentErrors=False, exitOnError=True)
-                self.writeTrx = lambda trxDataFile, block, blockNum: [trxDataFile.write(f"{trx['id']},{trx['block_num']},{trx['cpu_usage_us']},{trx['net_usage_words']}\n") for trx in block['payload']['transactions'] if block['payload']['transactions']]
+                self.writeTrx = lambda trxDataFile, block, blockNum:[ self.log_transactions(trxDataFile, block) ]
                 self.writeBlock = lambda blockDataFile, block: blockDataFile.write(f"{block['payload']['number']},{block['payload']['id']},{block['payload']['producer']},{block['payload']['status']},{block['payload']['timestamp']}\n")
                 self.fetchHeadBlock = lambda node, headBlock: node.processUrllibRequest("chain", "get_block_info", {"block_num":headBlock}, silentErrors=False, exitOnError=True)
 
@@ -159,6 +165,7 @@ def __init__(self, testHelperConfig: TestHelperConfig=TestHelperConfig(), cluste
         self.etcEosioLogsDirPath = self.etcLogsDirPath/Path("eosio")
         self.blockDataLogDirPath = self.loggingConfig.logDirPath/Path("blockDataLogs")
         self.blockDataPath = self.blockDataLogDirPath/Path("blockData.txt")
+        self.csvDataPath = self.blockDataLogDirPath/Path("csv.txt")
         self.blockTrxDataPath = self.blockDataLogDirPath/Path("blockTrxData.txt")
         self.reportPath = self.loggingConfig.logDirPath/Path("data.json")
 
@@ -385,7 +392,7 @@ def captureLowLevelArtifacts(self):
     def analyzeResultsAndReport(self, testResult: PtbTpsTestResult):
         args = self.prepArgs()
         artifactsLocate = log_reader.ArtifactPaths(nodeosLogPath=self.nodeosLogPath, trxGenLogDirPath=self.trxGenLogDirPath, blockTrxDataPath=self.blockTrxDataPath,
-                                                   blockDataPath=self.blockDataPath)
+                                                   blockDataPath=self.blockDataPath, csvDataPath=self.csvDataPath)
         tpsTestConfig = log_reader.TpsTestConfig(targetTps=self.ptbConfig.targetTps, testDurationSec=self.ptbConfig.testTrxGenDurationSec, tpsLimitPerGenerator=self.ptbConfig.tpsLimitPerGenerator,
                                                  numBlocksToPrune=self.ptbConfig.numAddlBlocksToPrune, numTrxGensUsed=testResult.numGeneratorsUsed,
                                                  targetTpsPerGenList=testResult.targetTpsPerGenList, quiet=self.ptbConfig.quiet)