set up for blob store

config.ts

@@ -60,13 +60,18 @@ const defaults = {
   ENABLE_MATCH_ARCHIVE: '', // Allow reading/writing parsed match blobs to S3 storage
   MATCH_ARCHIVE_S3_KEY_ID: '', // S3-compatible key ID to archive parsed match blobs
   MATCH_ARCHIVE_S3_KEY_SECRET: '', // S3-compatible key secret to archive parsed match blobs
-  MATCH_ARCHIVE_S3_ENDPOINT: '', // S3-compatible endpoint to archive parsed match blobs
+  MATCH_ARCHIVE_S3_ENDPOINT: '', // S3-compatible endpoint to archive parsed match blobs (should have http prefix)
   MATCH_ARCHIVE_S3_BUCKET: 'opendota', // name of the S3 bucket to archive parsed match blobs
   ENABLE_PLAYER_ARCHIVE: '', // Allow reading/writing player match blobs to S3 storage
   PLAYER_ARCHIVE_S3_KEY_ID: '', // S3-compatible key ID to archive player match blobs
   PLAYER_ARCHIVE_S3_KEY_SECRET: '', // S3-compatible key secret to archive player match blobs
   PLAYER_ARCHIVE_S3_ENDPOINT: '', // S3-compatible endpoint to archive player match blobs
-  PLAYER_ARCHIVE_S3_BUCKET: 'opendota-players', // name of the S3 bucket to archive player match blobs
+  PLAYER_ARCHIVE_S3_BUCKET: 'opendota-players', // name of the S3 bucket to archive player match blobs (should have http prefix)
+  BLOB_ARCHIVE_S3_KEY_ID: 'minioadmin', // S3-compatible key ID for match data blobs
+  BLOB_ARCHIVE_S3_KEY_SECRET: 'minioadmin', // S3-compatiable key secret for match data blobs
+  BLOB_ARCHIVE_S3_ENDPOINT: 'http://localhost:9000', // S3-compatible endpoint for match data blobs (should have http prefix)
+  BLOB_ARCHIVE_S3_BUCKET: 'opendota-blobs', // name of the S3 bucket to use for match data blobs
+  ENABLE_BLOB_ARCHIVE: '', // Allow reading match data blobs from S3 storage
   DISABLE_REPARSE: '', // Disable reparsing matches that are already parsed
   DISABLE_REPARSE_EARLY: '', // Disable reparsing matches by checking parsed_matches table first
   DISABLE_REGCDATA: '', // Disable refetching new GC data on every request (cache it)

dev/backfill.ts

@@ -0,0 +1,82 @@
+// Fetches old matches from Steam API and writes to blob storage
+import { Archive } from '../store/archive';
+import type { ApiMatch } from '../store/pgroup';
+import { generateJob, getSteamAPIData, transformMatch } from '../util/utility';
+import fs from 'fs';
+
+// following need to be set
+// STEAM_API_KEY needs to be set in .env
+// BLOB_ARCHIVE_S3_KEY_ID: 'minioadmin',
+// BLOB_ARCHIVE_S3_KEY_SECRET: 'minioadmin',
+// BLOB_ARCHIVE_S3_ENDPOINT: 'http://localhost:9000',
+// BLOB_ARCHIVE_S3_BUCKET: 'opendota-blobs',
+
+// This endpoint is limited to something like 1 request every 5 seconds
+const SCANNER_WAIT = 5000;
+const blobArchive = new Archive('blob');
+
+// We can stop at approximately 6400000000 (Feb 2024)
+async function scanApi() {
+  while (true) {
+    let nextSeqNum = Number(fs.readFileSync('./match_seq_num.txt')) || 0;
+    if (nextSeqNum > 6400000000) {
+      process.exit(0);
+    }
+    const container = generateJob('api_sequence', {
+      start_at_match_seq_num: nextSeqNum,
+    });
+    let data = null;
+    try {
+      data = await getSteamAPIData({
+        url: container.url,
+        // To use proxy we need to be running in the same environment as the proxies and STEAM_API_HOST env needs to be set
+        // proxy: true
+      });
+    } catch (err: any) {
+        console.log(err);
+        continue;
+      // unretryable steam error
+    //   if (err?.result?.status === 2) {
+    //     nextSeqNum += 1;
+    //     redisCount(redis, 'skip_seq_num');
+    //     // continue with next seq num
+    //     continue;
+    //   } else {
+    //     throw err;
+    //   }
+    }
+    const resp =
+      data && data.result && data.result.matches ? data.result.matches : [];
+    console.log('[API] match_seq_num:%s, matches:%s', nextSeqNum, resp.length);
+    // write to blobstore, process the blob using same function as insertMatch
+    try {
+    const insertResult = await Promise.all(resp.map(async (origMatch: ApiMatch) => {
+      const match = transformMatch(origMatch);
+      const result = await blobArchive.archivePut(match.match_id + '_api', Buffer.from(JSON.stringify(match)), true);
+      if (!result) {
+        throw new Error('failed to insert match ' + match.match_id);
+      }
+    }));
+    if (resp.length) {
+      nextSeqNum = resp[resp.length - 1].match_seq_num + 1;
+      console.log('next_seq_num: %s', nextSeqNum);
+      // Completed inserting matches on this page so update
+      fs.writeFileSync('./match_seq_num.txt', nextSeqNum.toString());
+    }
+  } catch (e) {
+    // If any fail, log the error and try the same number again
+    console.error(e);
+  }
+    await new Promise((resolve) =>
+      setTimeout(
+        resolve,
+        SCANNER_WAIT,
+      ),
+    );
+  }
+}
+
+async function start() {
+  await scanApi();
+}
+start();

docker-compose.override.yml

@@ -40,6 +40,18 @@ services:
   #       '-c',
   #       'sleep 10 && cqlsh -f /usr/src/sql/init.cql odota-scylla && cqlsh -f /usr/src/sql/create_tables.cql -k yasp odota-scylla',
   #     ]
+  minioinit:
+    image: minio/mc
+    depends_on:
+      - minio
+    volumes:
+      - .:/usr/src
+    container_name: odota-minioinit
+    entrypoint: >
+      /bin/sh -c "
+      /usr/bin/mc mb opendota-blobs;
+      exit 0;
+      "
   core:
     entrypoint: bash scripts/launch.sh
     volumes:

docker-compose.yml

@@ -16,6 +16,10 @@ services:
     container_name: odota-scylla
     command:
       - --skip-wait-for-gossip-to-settle 0
+  minio:
+    image: minio/minio:RELEASE.2024-11-07T00-52-20Z
+    container_name: odota-minio
+    command: server /data
   redis:
     image: redis:7
     container_name: odota-redis
@@ -52,11 +56,13 @@ services:
       CASSANDRA_URL: cassandra://odota-cassandra/yasp
       SCYLLA_URL: scylla://odota-scylla/yasp
       ELASTICSEARCH_URL: odota-elasticsearch:9200
+      BLOB_ARCHIVE_S3_ENDPOINT: http://odota-minio:9000
     links:
       - postgres
       - cassandra
       - redis
       - elasticsearch
       - parser
+      - minio
       # - scylla
     container_name: odota-core