diff --git a/.env.example b/.env.example index dfbd37c..61a0b6d 100644 --- a/.env.example +++ b/.env.example @@ -6,6 +6,8 @@ LISTEN_ADDR=:8080 KAFKA_BROKER=localhost:9092 ES_URL=http://localhost:9200 OLLAMA_HOST=http://localhost:11434 +OLLAMA_MODEL=llama3.2:1b +OLLAMA_REQUEST_TIMEOUT=30 MODEL_PATH=./isolation_forest.joblib # Dashboard diff --git a/docker-compose.yml b/docker-compose.yml index 134db3c..827cb42 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -90,6 +90,8 @@ services: KAFKA_BROKER: kafka:9092 ES_URL: http://elasticsearch:9200 OLLAMA_HOST: http://ollama:11434 + OLLAMA_MODEL: llama3.2:1b + OLLAMA_REQUEST_TIMEOUT: 30 MODEL_PATH: /app/data/isolation_forest.joblib depends_on: kafka: diff --git a/k8s/06-processing-service.yml b/k8s/06-processing-service.yml index 51c5ce7..7990f27 100644 --- a/k8s/06-processing-service.yml +++ b/k8s/06-processing-service.yml @@ -52,6 +52,10 @@ spec: value: "http://elasticsearch:9200" - name: OLLAMA_HOST value: "http://ollama:11434" + - name: OLLAMA_MODEL + value: "llama3.2:1b" + - name: OLLAMA_REQUEST_TIMEOUT + value: "30" resources: requests: memory: "512Mi" diff --git a/processing-service/processor.py b/processing-service/processor.py index ba194c0..d44a285 100644 --- a/processing-service/processor.py +++ b/processing-service/processor.py @@ -35,9 +35,14 @@ KAFKA_RECONNECT_BACKOFF_MS = 1000 KAFKA_RECONNECT_BACKOFF_MAX_MS = 60000 -# Ollama LLM client — host is configurable for K8s / Docker networking +# Ollama LLM client — host, model, and request timeout are all configurable +# for K8s / Docker networking and for swapping models without a code change. OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434") -_ollama_client = ollama.Client(host=OLLAMA_HOST) +OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3.2:1b") +# Bounds how long a single Ollama request may block, so a hung Ollama +# instance can't stall anomaly processing indefinitely. +OLLAMA_REQUEST_TIMEOUT = float(os.getenv("OLLAMA_REQUEST_TIMEOUT", "30")) +_ollama_client = ollama.Client(host=OLLAMA_HOST, timeout=OLLAMA_REQUEST_TIMEOUT) # --------------------------------------------------------------------------- @@ -123,7 +128,7 @@ def run_ai_analysis(log_text, log_data, last_failure_time): return True, "AI Analysis Unavailable (Ollama cooldown)", last_failure_time try: - response = _ollama_client.chat(model='llama3.2:1b', messages=[ + response = _ollama_client.chat(model=OLLAMA_MODEL, messages=[ { 'role': 'system', 'content': ( diff --git a/processing-service/test_processor.py b/processing-service/test_processor.py index 9f372a8..b48abb3 100644 --- a/processing-service/test_processor.py +++ b/processing-service/test_processor.py @@ -11,12 +11,14 @@ import pytest from sklearn.ensemble import IsolationForest +import processor from processor import ( KAFKA_RECONNECT_BACKOFF_MAX_MS, KAFKA_RECONNECT_BACKOFF_MS, KAFKA_RECONNECT_INITIAL_DELAY_SEC, KAFKA_RETRY_BACKOFF_MS, OLLAMA_COOLDOWN_SEC, + OLLAMA_MODEL, TRAIN_INTERVAL, check_rate_anomaly, has_keyword_indicator, @@ -179,6 +181,41 @@ def test_run_ai_analysis_ollama_exception(): assert new_failure > past_failure +def test_run_ai_analysis_uses_configured_model(): + """The chat() call must use OLLAMA_MODEL, not a hardcoded model name.""" + log_data = {"service": "test"} + past_failure = time.time() - OLLAMA_COOLDOWN_SEC - 10 + + with patch("processor._ollama_client.chat") as mock_chat: + mock_chat.return_value = {"message": {"content": "Normal"}} + run_ai_analysis("some info log", log_data, past_failure) + + assert mock_chat.call_args.kwargs["model"] == OLLAMA_MODEL + + +def test_ollama_model_env_var_overrides_default(monkeypatch): + """OLLAMA_MODEL is read from the environment at import time (with + "llama3.2:1b" as the default), and run_ai_analysis always uses the + current module-level value rather than a hardcoded literal - so + reconfiguring OLLAMA_MODEL changes the model used with no code change. + (monkeypatch.setattr on the already-imported module is used instead of + importlib.reload, because reloading processor.py re-executes its + module-level prometheus_client.Counter/Histogram registrations and + raises "Duplicated timeseries in CollectorRegistry".) + """ + assert processor.OLLAMA_MODEL == "llama3.2:1b" # unchanged default + + monkeypatch.setattr(processor, "OLLAMA_MODEL", "custom-model:latest") + + log_data = {"service": "test"} + past_failure = time.time() - OLLAMA_COOLDOWN_SEC - 10 + with patch("processor._ollama_client.chat") as mock_chat: + mock_chat.return_value = {"message": {"content": "Normal"}} + processor.run_ai_analysis("some info log", log_data, past_failure) + + assert mock_chat.call_args.kwargs["model"] == "custom-model:latest" + + def test_build_kafka_consumer_sets_backoff_config(): with patch("processor.KafkaConsumer") as mock_consumer: class EmptyConsumer: