Merge pull request #105 from LogicNet-Subnet/upload-log-minio

Upload validator pm2 log to Minio Storage

Author: LVH-Tony

.DS_Store

@@ -1,15 +1,20 @@
 module.exports = {
   apps: [{
-    name: "your_validator_name", // update this name
+    name: "sn35-validator", // update this name
     script: "neurons/validator/validator.py",
     interpreter: "/root/miniconda3/envs/sn35-env/bin/python", // update this path
     env: {
+        APP_NAME: "sn35-validator",
         PYTHONPATH: './:${PYTHONPATH}',
         OPENAI_API_KEY: "your_openai_key",
         USE_TORCH: 1,
         VALIDATOR_USERNAME: "datapool_username",
         VALIDATOR_PASSWORD: "datapool_password",
-        TASK_POOL_URL: "server_datapool_endpoint"
+        TASK_POOL_URL: "server_datapool_endpoint",
+        MINIO_ENDPOINT: "minio_endpoint",
+        MINIO_ACCESS_KEY: "minio_access_key",
+        MINIO_SECRET_KEY: "minio_secret_key",
+        PM2_LOG_DIR: "/root/.pm2/logs/"
     },
     args: [
         "--netuid", "35",

app_validator.config.sample.js

@@ -85,7 +85,6 @@ This setup allows you to run the Validator locally by hosting a vLLM server. Whi
 #### Prerequisites
 
 - **OpenAI API Key**: Obtain from the OpenAI platform dashboard.
-- **Wandb API Key**: Obtain from the Wandb platform dashboard.
 - **Python 3.10**
 - **PM2 Process Manager**: For running and managing the Validator process. *OPTIONAL*
 
@@ -117,6 +116,10 @@ This setup allows you to run the Validator locally by hosting a vLLM server. Whi
    echo "TASK_POOL_URL=server_datapool_endpoint"
    echo "VALIDATOR_USERNAME=datapool_username" >> .env
    echo "VALIDATOR_PASSWORD=datapool_account" >> .env
+   echo MINIO_ENDPOINT="server_minio_endpoint" >> .env
+   echo MINIO_ACCESS_KEY="minio_username" >> .env
+   echo MINIO_SECRET_KEY="minio_password" >> .env
+   echo APP_NAME="sn35-validator" >> .env
    echo "USE_TORCH=1" >> .env
    ```
 
@@ -157,26 +160,10 @@ This setup allows you to run the Validator locally by hosting a vLLM server. Whi
    ```bash
    --axon.port "your-public-open-port"
    ```
-
 ---
 
 ### Additional Features
 
-#### Wandb Integration
-
-Configure Wandb to track and analyze Validator performance.
-
-1. Add Wandb API key to `.env`:
-   ```bash
-   echo "WANDB_API_KEY=your_wandb_api_key" >> .env
-   ```
-2. It's already configured for mainnet as default.
-3. Run Validator with Wandb on Testnet:
-   ```bash
-   --wandb.project_name logicnet-testnet \
-   --wandb.entity ait-ai
-   ```
-
 ---
 
 ### Troubleshooting & Support

docs/VALIDATOR.md

@@ -6,6 +6,7 @@ pip install -e .
 # Uninstall uvloop if it's installed
 echo "Uninstalling uvloop..."
 pip uninstall uvloop -y
+bash pm2_setup.sh
 
 # Check if USE_TORCH=1 is already set in .env
 if grep -q '^USE_TORCH=1$' .env; then

install.sh

@@ -0,0 +1,61 @@
+from minio import Minio
+from minio.error import S3Error
+import os
+import time
+import glob
+
+class MinioManager:
+    def __init__(self, minio_endpoint, access_key, secret_key):
+        self.minio_endpoint = minio_endpoint
+        self.access_key = access_key
+        self.secret_key = secret_key
+        
+        self.minio_client = self.initialize_minio_client()
+
+    def initialize_minio_client(self):
+        """Initialize and return MinIO client."""
+        return Minio(
+            self.minio_endpoint,
+            access_key=self.access_key,
+            secret_key=self.secret_key,
+            secure=False  # Set to True if using HTTPS
+        )
+    
+    def ensure_bucket_exists(self, bucket_name):
+        """Check if bucket exists, create if it doesn't."""
+        try:
+            if not self.minio_client.bucket_exists(bucket_name):
+                self.minio_client.make_bucket(bucket_name)
+                print(f"Bucket '{bucket_name}' created")
+            else:
+                print(f"Bucket '{bucket_name}' already exists")
+        except S3Error as e:
+            print(f"Error checking/creating bucket: {e}")
+            raise
+
+    def upload_file(self, file_path, bucket_name, minio_folder_path):
+        """Upload a single file to MinIO."""
+        self.ensure_bucket_exists(bucket_name)
+        try:
+            object_name = os.path.basename(file_path)
+            if not os.path.exists(file_path):
+                print(f"File '{file_path}' not found, skipping")
+                return False
+            self.minio_client.fput_object(bucket_name, f"{minio_folder_path}/{object_name}", file_path)
+            print(f"Uploaded '{file_path}' to bucket '{bucket_name}' as '{object_name}'")
+            return True
+        except S3Error as e:
+            print(f"Error uploading '{file_path}': {e}")
+            return False
+        except Exception as e:
+            print(f"Unexpected error uploading '{file_path}': {e}")
+            return False
+        
+    def get_uploaded_files(self, bucket_name):
+        """Get list of files already uploaded to MinIO."""
+        try:
+            objects = self.minio_client.list_objects(bucket_name, recursive=True)
+            return {obj.object_name for obj in objects}
+        except S3Error as e:
+            print(f"Error listing objects in bucket: {e}")
+            return set()

logicnet/utils/minio_manager.py

@@ -15,19 +15,25 @@
 from copy import deepcopy
 import bittensor as bt
 import logicnet as ln
-from concurrent.futures import ThreadPoolExecutor, as_completed
 from neurons.validator.validator_proxy import ValidatorProxy
 from logicnet.base.validator import BaseValidatorNeuron
-from logicnet.validator import MinerManager, LogicChallenger, LogicRewarder, MinerInfo
+from logicnet.validator import MinerManager, LogicChallenger, LogicRewarder
 from logicnet.utils.wandb_manager import WandbManager
 from logicnet.utils.text_uts import modify_question
 from logicnet.protocol import LogicSynapse
 from neurons.validator.core.serving_queue import QueryQueue
-from collections import defaultdict
-import wandb
 from threading import Lock
 import queue
+from logicnet.utils.minio_manager import MinioManager
+import glob
 
+log_bucket_name = "logs"
+app_name = os.getenv("APP_NAME", "validator")
+validator_username = os.getenv("VALIDATOR_USERNAME")
+minio_endpoint = os.getenv("MINIO_ENDPOINT")
+access_key = os.getenv("MINIO_ACCESS_KEY")
+secret_key = os.getenv("MINIO_SECRET_KEY")
+pm2_log_dir = os.getenv("PM2_LOG_DIR", "/root/.pm2/logs")
 
 def init_category(config=None, model_pool=None):
     category = {
@@ -50,6 +56,14 @@ def init_category(config=None, model_pool=None):
     "mistralai/Mistral-7B-Instruct"
 ]
 
+def get_latest_previous_log_file(log_files):
+    """Return the second-most-recent log file based on modification time."""
+    if len(log_files) < 2:
+        return None  # Not enough files to have a "previous" file
+    # Sort files by modification time (most recent first)
+    sorted_files = sorted(log_files, key=lambda x: os.path.getmtime(x), reverse=True)
+    return sorted_files[1]  # Second file is the latest previous
+
 class Validator(BaseValidatorNeuron):
     def __init__(self, config=None):
         """
@@ -568,10 +582,57 @@ def _log_wandb(self, log):
         except Exception as e:
             bt.logging.error(f"Error logging to wandb: {e}")
 
-
 # The main function parses the configuration and runs the validator.
 if __name__ == "__main__":
+    last_err_file_name = ""
+    last_out_file_name = ""
+    
+    try:
+        minio_manager = MinioManager(minio_endpoint, access_key, secret_key)
+    except Exception as e:
+        bt.logging.error(f"Error initializing MinioManager: {e}")
+    
     with Validator() as validator:
         while True:
             bt.logging.info("\033[1;32m🟢 Validator running...\033[0m", time.time())
-            time.sleep(360)
+
+            #########################################################
+            # UPLOAD OUT LOG FILES
+            out_log_files = glob.glob(os.path.join(pm2_log_dir, f"*{app_name}-out*.log"))
+            # bt.logging.info(out_log_files)
+
+            current_file_count = len(out_log_files)
+            # Detect rotation (new file added)
+            if current_file_count >= 2:
+                # A new file was created, so upload the latest previous file
+                previous_file = get_latest_previous_log_file(out_log_files)
+                if previous_file != last_out_file_name and previous_file:
+                    last_out_file_name = previous_file
+                    file_name = os.path.basename(previous_file)
+                    if file_name not in minio_manager.get_uploaded_files(log_bucket_name):
+                        bt.logging.info(f"Uploading {previous_file} to MinIO")
+                        if minio_manager.upload_file(previous_file, log_bucket_name, validator_username):
+                            bt.logging.info(f"\033[1;32m✅ Uploaded {file_name} to MinIO\033[0m")
+            #########################################################
+
+
+            #########################################################
+            # UPLOAD ERR LOG FILES
+            err_log_files = glob.glob(os.path.join(pm2_log_dir, f"*{app_name}-error*.log"))
+            # bt.logging.info(err_log_files)
+            current_file_count = len(err_log_files)
+
+            # Detect rotation (new file added)
+            if current_file_count >= 2:
+                # A new file was created, so upload the latest previous file
+                previous_file = get_latest_previous_log_file(err_log_files)
+                if previous_file != last_err_file_name and previous_file:
+                    last_err_file_name = previous_file
+                    file_name = os.path.basename(previous_file)
+                    if file_name not in minio_manager.get_uploaded_files(log_bucket_name):
+                        bt.logging.info(f"Uploading {previous_file} to MinIO")
+                        if minio_manager.upload_file(previous_file, log_bucket_name, validator_username):
+                            bt.logging.info(f"\033[1;32m✅ Uploaded {file_name} to MinIO\033[0m")
+            #########################################################
+
+            time.sleep(60)

neurons/validator/validator.py

@@ -0,0 +1,7 @@
+pm2 install pm2-logrotate
+
+pm2 set pm2-logrotate:max_size 10M
+pm2 set pm2-logrotate:compress false
+# Rotate every 10 minutes
+pm2 set pm2-logrotate:rotateInterval '*/10 * * * *'  
+pm2 set pm2-logrotate:retain 100

pm2_setup.sh

@@ -13,4 +13,5 @@ python-dotenv==1.0.1
 loguru
 sympy
 wandb
-datasets
+datasets
+minio