fix: refactor online evals (#12654)

move online evals to enterprise repo
This commit is contained in:
udev
2026-06-15 10:13:55 +00:00
committed by GitHub
parent 291903345e
commit 6761ce2dbf
30 changed files with 249 additions and 4795 deletions
+41 -5
View File
@@ -5,9 +5,10 @@
"version": "0.2.0",
"configurations": [
{
"type": "lldb",
"type": "codelldb",
"request": "launch",
"name": "Debug executable 'openobserve' (server)",
"program": "${env:HOME}/.cargo/shared-target/debug/openobserve",
"cargo": {
"args": ["build", "--bin=openobserve", "--package=openobserve"],
"filter": {
@@ -17,15 +18,16 @@
},
"env": {
"ZO_ROOT_USER_EMAIL": "root@example.com",
"ZO_ROOT_USER_PASSWORD": "Complexpass#123",
"ZO_ROOT_USER_PASSWORD": "Complexpass#123"
},
"args": [],
"cwd": "${workspaceFolder}"
},
{
"type": "lldb",
"type": "codelldb",
"request": "launch",
"name": "Debug unit tests in library 'openobserve'",
"program": "${cargo:program}",
"cargo": {
"args": ["test", "--no-run", "--lib", "--package=openobserve"],
"filter": {
@@ -37,9 +39,10 @@
"cwd": "${workspaceFolder}"
},
{
"type": "lldb",
"type": "codelldb",
"request": "launch",
"name": "Debug executable 'openobserve' (bin)",
"program": "${env:HOME}/.cargo/shared-target/debug/openobserve",
"cargo": {
"args": ["build", "--bin=openobserve", "--package=openobserve"],
"filter": {
@@ -51,9 +54,10 @@
"cwd": "${workspaceFolder}"
},
{
"type": "lldb",
"type": "codelldb",
"request": "launch",
"name": "Debug unit tests in executable 'openobserve'",
"program": "${cargo:program}",
"cargo": {
"args": [
"test",
@@ -68,6 +72,38 @@
},
"args": [],
"cwd": "${workspaceFolder}"
},
{
"type": "codelldb",
"request": "launch",
"name": "Debug executable 'openobserve' (enterprise)",
"program": "${env:HOME}/.cargo/shared-target/debug/openobserve",
"cargo": {
"args": ["build", "--bin=openobserve", "--package=openobserve"],
"filter": {
"name": "openobserve",
"kind": "bin"
}
},
"envFile": "${workspaceFolder}/.env",
"args": [],
"cwd": "${workspaceFolder}"
},
{
"type": "codelldb",
"request": "launch",
"name": "Debug executable 'openobserve' (cloud)",
"program": "${env:HOME}/.cargo/shared-target/debug/openobserve",
"cargo": {
"args": ["build", "--bin=openobserve", "--package=openobserve"],
"filter": {
"name": "openobserve",
"kind": "bin"
}
},
"envFile": "${workspaceFolder}/.env.cloud",
"args": [],
"cwd": "${workspaceFolder}"
}
]
}
+1
View File
@@ -109,6 +109,7 @@ pub fn is_ofga_unsupported(name: &str) -> bool {
RE_OFGA_UNSUPPORTED_NAME.is_match(name)
}
#[cfg(feature = "enterprise")]
pub(crate) fn is_ofga_object_visible(
org_id: &str,
object_type: &str,
-6
View File
@@ -1364,12 +1364,6 @@ pub struct Common {
help = "Enable user-defined model pricing. When true, uses DB pricing definitions and syncs from GitHub. When false, falls back to hardcoded built-in pricing only."
)]
pub model_pricing_enabled: bool,
#[env_config(
name = "ZO_ONLINE_EVALS_ENABLED",
default = true,
help = "Show the Online Evaluations UI (top-level Evaluations route) and the LLM Providers Settings page. When false, both are hidden. The backend endpoints remain reachable regardless — this flag only gates the frontend surface."
)]
pub online_evals_enabled: bool,
#[env_config(
name = "ZO_MODEL_PRICING_SOURCE_URL",
default = "https://raw.githubusercontent.com/openobserve/sdr_patterns/refs/heads/main/llm_pricing.json",
+4
View File
@@ -24,10 +24,14 @@ pub mod alerts;
pub mod billings;
pub mod dashboards;
pub mod destinations;
#[cfg(feature = "enterprise")]
pub mod eval_jobs;
pub mod folders;
pub mod pipelines;
#[cfg(feature = "enterprise")]
pub mod providers;
pub mod reports;
#[cfg(feature = "enterprise")]
pub mod score_configs;
#[cfg(feature = "enterprise")]
pub mod scorers;
+8 -2
View File
@@ -34,9 +34,15 @@ use crate::{
impl From<EvalJobError> for Response {
fn from(value: EvalJobError) -> Self {
match value {
EvalJobError::InfraError(err) => MetaHttpResponse::internal_error(err),
EvalJobError::InfraError(err) => {
log::error!("[EvalJob] internal error: {err}");
MetaHttpResponse::internal_error("Internal server error")
}
EvalJobError::NotFound => MetaHttpResponse::not_found(value),
EvalJobError::ReconcilerError(_) => MetaHttpResponse::internal_error(value),
EvalJobError::ReconcilerError(err) => {
log::error!("[EvalJob] reconciler error: {err}");
MetaHttpResponse::internal_error("Internal server error")
}
EvalJobError::InvalidStatus(_) | EvalJobError::InvalidStatusTransition { .. } => {
MetaHttpResponse::bad_request(value)
}
+4
View File
@@ -32,6 +32,7 @@ pub mod dashboards;
#[cfg(feature = "enterprise")]
pub mod domain_management;
pub mod enrichment_table;
#[cfg(feature = "enterprise")]
pub mod eval_jobs;
#[allow(deprecated)]
pub mod folders;
@@ -51,12 +52,15 @@ pub mod pipelines;
#[cfg(feature = "profiling")]
pub mod profiling;
pub mod promql;
#[cfg(feature = "enterprise")]
pub mod providers;
pub mod ratelimit;
#[cfg(feature = "enterprise")]
pub mod re_pattern;
pub mod rum;
#[cfg(feature = "enterprise")]
pub mod score_configs;
#[cfg(feature = "enterprise")]
pub mod scorers;
pub mod search;
pub mod service_accounts;
+4 -1
View File
@@ -30,7 +30,10 @@ use crate::{
impl From<ProviderError> for Response {
fn from(value: ProviderError) -> Self {
match value {
ProviderError::InfraError(err) => MetaHttpResponse::internal_error(err),
ProviderError::InfraError(err) => {
log::error!("[Provider] internal error: {err}");
MetaHttpResponse::internal_error("Internal server error")
}
ProviderError::MissingName => {
MetaHttpResponse::bad_request("Provider name cannot be empty")
}
+42 -26
View File
@@ -20,7 +20,10 @@ use crate::common::utils::auth::UserEmail;
#[cfg(feature = "enterprise")]
use crate::handler::http::extractors::Headers;
use crate::{
common::meta::http::HttpResponse as MetaHttpResponse,
common::{
meta::{authz::Authz, http::HttpResponse as MetaHttpResponse},
utils::auth::{is_ofga_object_visible, remove_ownership, set_ownership},
},
handler::http::models::score_configs::{
ListScoreConfigVersionsResponseBody, ListScoreConfigsResponseBody, ScoreConfigRequestBody,
ScoreConfigResponseBody, ScoreConfigUpdateRequestBody,
@@ -28,19 +31,20 @@ use crate::{
service::llm_evaluations::score_configs::{self, ScoreConfigError},
};
impl From<ScoreConfigError> for Response {
fn from(value: ScoreConfigError) -> Self {
match value {
ScoreConfigError::InfraError(err) => MetaHttpResponse::internal_error(err),
ScoreConfigError::MissingName => {
MetaHttpResponse::bad_request("Score config name cannot be empty")
}
ScoreConfigError::NotFound => MetaHttpResponse::not_found("Score config not found"),
ScoreConfigError::DuplicateName => {
MetaHttpResponse::conflict("Score config name already exists")
}
ScoreConfigError::InUseByScorer => MetaHttpResponse::conflict(value),
fn score_config_error_response(value: ScoreConfigError) -> Response {
match value {
ScoreConfigError::InfraError(err) => {
log::error!("[ScoreConfig] internal error: {err}");
MetaHttpResponse::internal_error("Internal server error")
}
ScoreConfigError::MissingName => {
MetaHttpResponse::bad_request("Score config name cannot be empty")
}
ScoreConfigError::NotFound => MetaHttpResponse::not_found("Score config not found"),
ScoreConfigError::DuplicateName => {
MetaHttpResponse::conflict("Score config name already exists")
}
ScoreConfigError::InUseByScorer => MetaHttpResponse::conflict(value),
}
}
@@ -80,15 +84,23 @@ pub async fn list_score_configs(
Err(e) => return MetaHttpResponse::forbidden(e.to_string()),
}
};
#[cfg(not(feature = "enterprise"))]
let permitted_objects = None;
match score_configs::list_score_configs(&org_id, permitted_objects).await {
match score_configs::list_score_configs(&org_id).await {
Ok(list) => {
let list: Vec<infra::table::score_configs::ScoreConfig> = list
.into_iter()
.filter(|config| {
is_ofga_object_visible(
&org_id,
"score_config",
&config.entity_id,
permitted_objects.as_deref(),
)
})
.collect();
let body: ListScoreConfigsResponseBody = list.into();
MetaHttpResponse::json(body)
}
Err(err) => err.into(),
Err(err) => score_config_error_response(err),
}
}
@@ -120,10 +132,11 @@ pub async fn create_score_config(
let config: infra::table::score_configs::ScoreConfig = body.into();
match score_configs::save_score_config(&org_id, config).await {
Ok(c) => {
set_ownership(&org_id, "score_configs", Authz::new(&c.entity_id)).await;
let resp: ScoreConfigResponseBody = c.into();
MetaHttpResponse::json(resp)
}
Err(err) => err.into(),
Err(err) => score_config_error_response(err),
}
}
@@ -155,7 +168,7 @@ pub async fn get_score_config(Path((org_id, entity_id)): Path<(String, String)>)
let resp: ScoreConfigResponseBody = c.into();
MetaHttpResponse::json(resp)
}
Err(err) => err.into(),
Err(err) => score_config_error_response(err),
}
}
@@ -188,7 +201,7 @@ pub async fn list_score_config_versions(
let body: ListScoreConfigVersionsResponseBody = versions.into();
MetaHttpResponse::json(body)
}
Err(err) => err.into(),
Err(err) => score_config_error_response(err),
}
}
@@ -225,7 +238,7 @@ pub async fn update_score_config(
let resp: ScoreConfigResponseBody = c.into();
MetaHttpResponse::json(resp)
}
Err(err) => err.into(),
Err(err) => score_config_error_response(err),
}
}
@@ -254,8 +267,11 @@ pub async fn update_score_config(
)]
pub async fn delete_score_config(Path((org_id, entity_id)): Path<(String, String)>) -> Response {
match score_configs::delete_score_config(&org_id, &entity_id).await {
Ok(()) => MetaHttpResponse::ok("Score config deactivated"),
Err(err) => err.into(),
Ok(()) => {
remove_ownership(&org_id, "score_configs", Authz::new(&entity_id)).await;
MetaHttpResponse::ok("Score config deactivated")
}
Err(err) => score_config_error_response(err),
}
}
@@ -272,7 +288,7 @@ mod tests {
(ScoreConfigError::InUseByScorer, 409),
];
for (err, expected) in cases {
let resp: Response = err.into();
let resp = score_config_error_response(err);
assert_eq!(resp.status().as_u16(), expected);
}
}
@@ -280,7 +296,7 @@ mod tests {
#[test]
fn test_score_config_error_infra_is_500() {
let err = ScoreConfigError::InfraError(infra::errors::Error::Message("db".to_string()));
let resp: Response = err.into();
let resp = score_config_error_response(err);
assert_eq!(resp.status().as_u16(), 500);
}
}
+45 -31
View File
@@ -23,7 +23,10 @@ use crate::common::utils::auth::UserEmail;
#[cfg(feature = "enterprise")]
use crate::handler::http::extractors::Headers;
use crate::{
common::meta::http::HttpResponse as MetaHttpResponse,
common::{
meta::{authz::Authz, http::HttpResponse as MetaHttpResponse},
utils::auth::{is_ofga_object_visible, remove_ownership, set_ownership},
},
handler::http::models::scorers::{
ListScorerVersionsResponseBody, ListScorersQuery, ListScorersResponseBody,
LlmJudgeOutputSchemaRequestBody, LlmJudgeOutputSchemaResponseBody, ScorerRequestBody,
@@ -36,22 +39,21 @@ use crate::{
},
};
impl From<ScorerError> for Response {
fn from(value: ScorerError) -> Self {
match value {
ScorerError::InfraError(err) => MetaHttpResponse::internal_error(err),
ScorerError::MissingName => {
MetaHttpResponse::bad_request("Scorer name cannot be empty")
}
ScorerError::NotFound => MetaHttpResponse::not_found("Scorer not found"),
ScorerError::InvalidScorerType(_)
| ScorerError::ScorerTypeImmutable
| ScorerError::ProducesScoreConfigIdImmutable
| ScorerError::ScoreConfigVersionNotFound
| ScorerError::InvalidOutputSchema(_) => MetaHttpResponse::bad_request(value),
ScorerError::DuplicateName => MetaHttpResponse::conflict(value),
ScorerError::InUseByEvalJob => MetaHttpResponse::conflict(value),
fn scorer_error_response(value: ScorerError) -> Response {
match value {
ScorerError::InfraError(err) => {
log::error!("[Scorer] internal error: {err}");
MetaHttpResponse::internal_error("Internal server error")
}
ScorerError::MissingName => MetaHttpResponse::bad_request("Scorer name cannot be empty"),
ScorerError::NotFound => MetaHttpResponse::not_found("Scorer not found"),
ScorerError::InvalidScorerType(_)
| ScorerError::ScorerTypeImmutable
| ScorerError::ProducesScoreConfigIdImmutable
| ScorerError::ScoreConfigVersionNotFound
| ScorerError::InvalidOutputSchema(_) => MetaHttpResponse::bad_request(value),
ScorerError::DuplicateName => MetaHttpResponse::conflict(value),
ScorerError::InUseByEvalJob => MetaHttpResponse::conflict(value),
}
}
@@ -95,15 +97,23 @@ pub async fn list_scorers(
Err(e) => return MetaHttpResponse::forbidden(e.to_string()),
}
};
#[cfg(not(feature = "enterprise"))]
let permitted_objects = None;
match scorers::list_scorers(&org_id, query.scorer_type.as_ref(), permitted_objects).await {
match scorers::list_scorers(&org_id, query.scorer_type.as_ref()).await {
Ok(list) => {
let list: Vec<infra::table::scorers::Scorer> = list
.into_iter()
.filter(|scorer| {
is_ofga_object_visible(
&org_id,
"scorer",
&scorer.entity_id,
permitted_objects.as_deref(),
)
})
.collect();
let body: ListScorersResponseBody = list.into();
MetaHttpResponse::json(body)
}
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -135,10 +145,11 @@ pub async fn create_scorer(
let scorer: infra::table::scorers::Scorer = body.into();
match scorers::save_scorer(&org_id, scorer).await {
Ok(s) => {
set_ownership(&org_id, "scorers", Authz::new(&s.entity_id)).await;
let resp: ScorerResponseBody = s.into();
MetaHttpResponse::json(resp)
}
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -170,7 +181,7 @@ pub async fn get_scorer(Path((org_id, entity_id)): Path<(String, String)>) -> Re
let resp: ScorerResponseBody = s.into();
MetaHttpResponse::json(resp)
}
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -201,7 +212,7 @@ pub async fn list_scorer_versions(Path((org_id, entity_id)): Path<(String, Strin
let body: ListScorerVersionsResponseBody = versions.into();
MetaHttpResponse::json(body)
}
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -238,7 +249,7 @@ pub async fn update_scorer(
let resp: ScorerResponseBody = s.into();
MetaHttpResponse::json(resp)
}
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -267,8 +278,11 @@ pub async fn update_scorer(
)]
pub async fn delete_scorer(Path((org_id, entity_id)): Path<(String, String)>) -> Response {
match scorers::delete_scorer(&org_id, &entity_id).await {
Ok(()) => MetaHttpResponse::ok("Scorer deactivated"),
Err(err) => err.into(),
Ok(()) => {
remove_ownership(&org_id, "scorers", Authz::new(&entity_id)).await;
MetaHttpResponse::ok("Scorer deactivated")
}
Err(err) => scorer_error_response(err),
}
}
@@ -299,7 +313,7 @@ pub async fn test_scorer(
) -> Response {
match run_scorer_test(&org_id, &body).await {
Ok(resp) => MetaHttpResponse::json(resp),
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -331,7 +345,7 @@ pub async fn preview_llm_judge_output_schema(
Ok(output_schema) => {
MetaHttpResponse::json(LlmJudgeOutputSchemaResponseBody { output_schema })
}
Err(err) => err.into(),
Err(err) => scorer_error_response(err),
}
}
@@ -548,7 +562,7 @@ mod tests {
(ScorerError::InUseByEvalJob, 409),
];
for (err, expected) in cases {
let resp: Response = err.into();
let resp = scorer_error_response(err);
assert_eq!(resp.status().as_u16(), expected);
}
}
@@ -556,7 +570,7 @@ mod tests {
#[test]
fn test_scorer_error_infra_is_500() {
let err = ScorerError::InfraError(infra::errors::Error::Message("db".to_string()));
let resp: Response = err.into();
let resp = scorer_error_response(err);
assert_eq!(resp.status().as_u16(), 500);
}
+2 -1
View File
@@ -366,6 +366,7 @@ pub async fn zo_config() -> impl IntoResponse {
let service_streams_enabled = enterprise_value!(false, o2cfg.service_streams.enabled);
// Anomaly detection is always on when the enterprise feature is compiled in — no runtime flag.
let anomaly_detection_enabled = enterprise_value!(false, true);
let online_evals_enabled = enterprise_value!(false, o2cfg.common.online_evals_enabled);
#[cfg(all(feature = "cloud", not(feature = "enterprise")))]
let build_type = "cloud";
@@ -473,7 +474,7 @@ pub async fn zo_config() -> impl IntoResponse {
incidents_enabled,
service_streams_enabled,
model_pricing_enabled: cfg.common.model_pricing_enabled,
online_evals_enabled: cfg.common.online_evals_enabled,
online_evals_enabled,
anomaly_detection_enabled,
enable_cross_linking: cfg.common.enable_cross_linking,
show_fts_field_values: cfg.common.show_fts_field_values,
+31 -25
View File
@@ -850,35 +850,41 @@ pub fn service_routes() -> Router {
// sourcemaps
.route("/{org_id}/sourcemaps",get(sourcemaps::list).post(sourcemaps::upload_maps).delete(sourcemaps::delete))
.route("/{org_id}/sourcemaps/values",get(sourcemaps::list_values))
.route("/{org_id}/sourcemaps/stacktrace",post(sourcemaps::translate_stacktrace))
.route("/{org_id}/sourcemaps/stacktrace",post(sourcemaps::translate_stacktrace));
// LLM Providers (Online Eval Phase 2)
.route("/{org_id}/providers", get(providers::list_providers).post(providers::create_provider))
.route("/{org_id}/providers/{provider_id}", get(providers::get_provider).put(providers::update_provider).delete(providers::delete_provider))
.route("/{org_id}/providers/{provider_id}/test", post(providers::test_provider))
#[cfg(feature = "enterprise")]
{
if get_o2_config().common.online_evals_enabled {
router = router
// LLM Providers (Online Eval Phase 2)
.route("/{org_id}/providers", get(providers::list_providers).post(providers::create_provider))
.route("/{org_id}/providers/{provider_id}", get(providers::get_provider).put(providers::update_provider).delete(providers::delete_provider))
.route("/{org_id}/providers/{provider_id}/test", post(providers::test_provider))
// Score Configs (Online Eval Phase 2)
// NOTE: /{entity_id}/versions must precede /{entity_id} for routing correctness
.route("/{org_id}/score_configs", get(score_configs::list_score_configs).post(score_configs::create_score_config))
.route("/{org_id}/score_configs/{entity_id}/versions", get(score_configs::list_score_config_versions))
.route("/{org_id}/score_configs/{entity_id}", get(score_configs::get_score_config).put(score_configs::update_score_config).delete(score_configs::delete_score_config))
// Score Configs (Online Eval Phase 2)
// NOTE: /{entity_id}/versions must precede /{entity_id} for routing correctness
.route("/{org_id}/score_configs", get(score_configs::list_score_configs).post(score_configs::create_score_config))
.route("/{org_id}/score_configs/{entity_id}/versions", get(score_configs::list_score_config_versions))
.route("/{org_id}/score_configs/{entity_id}", get(score_configs::get_score_config).put(score_configs::update_score_config).delete(score_configs::delete_score_config))
// Scorers (Online Eval Phase 2)
// NOTE: static and nested routes must precede /{entity_id}
.route("/{org_id}/scorers", get(scorers::list_scorers).post(scorers::create_scorer))
.route("/{org_id}/scorers/test", post(scorers::test_scorer))
.route("/{org_id}/scorers/llm_judge/output_schema", post(scorers::preview_llm_judge_output_schema))
.route("/{org_id}/scorers/{entity_id}/versions", get(scorers::list_scorer_versions))
.route("/{org_id}/scorers/{entity_id}", get(scorers::get_scorer).put(scorers::update_scorer).delete(scorers::delete_scorer))
// Scorers (Online Eval Phase 2)
// NOTE: static and nested routes must precede /{entity_id}
.route("/{org_id}/scorers", get(scorers::list_scorers).post(scorers::create_scorer))
.route("/{org_id}/scorers/test", post(scorers::test_scorer))
.route("/{org_id}/scorers/llm_judge/output_schema", post(scorers::preview_llm_judge_output_schema))
.route("/{org_id}/scorers/{entity_id}/versions", get(scorers::list_scorer_versions))
.route("/{org_id}/scorers/{entity_id}", get(scorers::get_scorer).put(scorers::update_scorer).delete(scorers::delete_scorer))
// Online Eval Jobs (Online Eval Phase 2)
// NOTE: /activate, /pause, /resume, /archive must precede /{job_id}
.route("/{org_id}/eval_jobs", get(eval_jobs::list_eval_jobs).post(eval_jobs::create_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/activate", post(eval_jobs::activate_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/pause", post(eval_jobs::pause_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/resume", post(eval_jobs::resume_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/archive", post(eval_jobs::archive_eval_job))
.route("/{org_id}/eval_jobs/{job_id}", get(eval_jobs::get_eval_job).put(eval_jobs::update_eval_job).delete(eval_jobs::delete_eval_job));
// Online Eval Jobs (Online Eval Phase 2)
// NOTE: /activate, /pause, /resume, /archive must precede /{job_id}
.route("/{org_id}/eval_jobs", get(eval_jobs::list_eval_jobs).post(eval_jobs::create_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/activate", post(eval_jobs::activate_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/pause", post(eval_jobs::pause_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/resume", post(eval_jobs::resume_eval_job))
.route("/{org_id}/eval_jobs/{job_id}/archive", post(eval_jobs::archive_eval_job))
.route("/{org_id}/eval_jobs/{job_id}", get(eval_jobs::get_eval_job).put(eval_jobs::update_eval_job).delete(eval_jobs::delete_eval_job));
}
}
#[cfg(feature = "enterprise")]
{
+21 -1
View File
@@ -40,7 +40,7 @@ use crate::{
pub enum PipelineError {
// internal
#[error("InfraError# {0}")]
InfraError(#[from] infra::errors::Error),
InfraError(infra::errors::Error),
// not found
#[error("Pipeline with ID {0} not found.")]
NotFound(String),
@@ -60,6 +60,17 @@ pub enum PipelineError {
DeleteDerivedStream(String),
}
impl From<infra::errors::Error> for PipelineError {
fn from(value: infra::errors::Error) -> Self {
match value {
infra::errors::Error::DbError(infra::errors::DbError::KeyNotExists(key)) => {
PipelineError::NotFound(key)
}
err => PipelineError::InfraError(err),
}
}
}
/// Stores a new pipeline to database.
///
/// Pipeline validation should be handled by the caller.
@@ -631,4 +642,13 @@ mod tests {
// Clean up
remove_scheduled_pipeline_from_cache(scheduled_id).await;
}
#[test]
fn test_key_not_exists_maps_to_not_found() {
let err = PipelineError::from(infra::errors::Error::DbError(
infra::errors::DbError::KeyNotExists("0".to_string()),
));
assert!(matches!(err, PipelineError::NotFound(id) if id == "0"));
}
}
@@ -1,496 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Executor Runtime — orchestrates span evaluation through configured scorers.
//!
//! Glue between the pipeline's LlmEvaluation node and the individual scorer
//! executors (llm_judge, remote). Called from `batch_execution.rs` when the
//! pipeline node is an evaluation pipeline (job_id is set).
//!
//! For each span record:
//! 1. Resolves scorer configs from the database
//! 2. Dispatches to llm_judge or remote scorer
//! 3. Produces score records (forwarded through pipeline to _llm_scores)
//! 4. Produces evaluator traces (internally exported as OTLP spans)
use std::collections::HashMap;
use anyhow::Result;
use config::meta::{pipeline::components::ScorerRef, self_reporting::evaluator};
use infra::table::{online_eval_jobs::JobInputMapping, scorers::ScorerType};
use serde_json::Value;
use crate::service::llm_evaluations::{
evaluator_trace::{EvaluatorTrace, EvaluatorTraceInput, create_evaluator_trace},
prepared_scorers::{PreparedScorer, ScorerExecutionError},
};
#[derive(Debug)]
pub struct SpanEvalContext {
pub org_id: String,
pub span_id: String,
pub trace_id: String,
pub evaluator_trace_id: String,
pub session_id: Option<String>,
pub source_stream: String,
pub job_id: Option<String>,
pub eval_run_id: Option<String>,
pub sampling_rate: Option<f64>,
pub sampled: Option<bool>,
pub attributes: HashMap<String, Value>,
}
#[derive(Debug)]
pub struct EvalError {
pub scorer_id: String,
pub scorer_type: Option<ScorerType>,
pub error_kind: String,
pub error_message: String,
pub latency_ms: i64,
}
pub struct ExecutorOutput {
pub scores: Vec<Value>,
pub evaluator_traces: Vec<EvaluatorTrace>,
pub errors: Vec<EvalError>,
}
impl ExecutorOutput {
pub fn empty() -> Self {
Self {
scores: Vec::new(),
evaluator_traces: Vec::new(),
errors: Vec::new(),
}
}
}
pub fn extract_context_from_span(
org_id: &str,
job_id: Option<&str>,
record: &Value,
) -> Option<SpanEvalContext> {
let span_id = record
.get("span_id")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.unwrap_or_default();
let trace_id = record
.get("trace_id")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.unwrap_or_default();
if span_id.is_empty() {
log::warn!("[EXECUTOR-RUNTIME] Skipping evaluation span with missing span_id");
return None;
}
if trace_id.is_empty() {
log::warn!(
"[EXECUTOR-RUNTIME] Skipping evaluation span '{}' with missing trace_id",
span_id
);
return None;
}
let session_id = record
.get("session_id")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let source_stream = record
.get("pipeline_source_stream")
.and_then(|v| v.as_str())
.unwrap_or("traces")
.to_string();
let mut attributes: HashMap<String, Value> = HashMap::new();
if let Some(attrs) = record.get("attributes").and_then(|v| v.as_object()) {
for (k, v) in attrs {
attributes.insert(k.clone(), v.clone());
}
}
for (key, value) in record.as_object().into_iter().flat_map(|o| o.iter()) {
if key == "span_id"
|| key == "trace_id"
|| key == "session_id"
|| key == "attributes"
|| key == "pipeline_source_stream"
{
continue;
}
attributes
.entry(key.clone())
.or_insert_with(|| value.clone());
}
Some(SpanEvalContext {
org_id: org_id.to_string(),
span_id,
trace_id,
evaluator_trace_id: config::ider::generate_trace_id(),
session_id,
source_stream,
job_id: job_id.map(|s| s.to_string()),
eval_run_id: None,
sampling_rate: None,
sampled: None,
attributes,
})
}
fn resolve_scorer_input_variables(
job_input_mapping: &Option<JobInputMapping>,
scorer_entity_id: &str,
span_attributes: &HashMap<String, Value>,
) -> HashMap<String, Value> {
let mut input_variables = HashMap::new();
let Some(mapping) = job_input_mapping
.as_ref()
.and_then(|m| m.get(scorer_entity_id))
else {
return input_variables;
};
for (variable, pattern) in mapping {
input_variables.insert(
variable.clone(),
Value::String(render_mapping_value(pattern, span_attributes)),
);
}
input_variables
}
fn render_mapping_value(pattern: &str, attrs: &HashMap<String, Value>) -> String {
if pattern.starts_with("{{") && pattern.ends_with("}}") && pattern.len() > 4 {
let path = pattern[2..pattern.len() - 2].trim();
return get_attr(attrs, path);
}
pattern.to_string()
}
fn get_attr(attrs: &HashMap<String, Value>, path: &str) -> String {
let parts: Vec<&str> = path.split('.').collect();
let mut current = attrs.get(parts[0]);
for part in &parts[1..] {
current = current.and_then(|v| v.get(part));
}
match current {
Some(Value::String(s)) => s.clone(),
Some(v) => v.to_string(),
None => attrs
.get(path)
.map(|v| match v {
Value::String(s) => s.clone(),
v => v.to_string(),
})
.unwrap_or_default(),
}
}
fn input_mapping_attributes(ctx: &SpanEvalContext) -> HashMap<String, Value> {
let mut attrs = ctx.attributes.clone();
attrs.insert("span_id".to_string(), Value::String(ctx.span_id.clone()));
attrs.insert("trace_id".to_string(), Value::String(ctx.trace_id.clone()));
attrs.insert(
"pipeline_source_stream".to_string(),
Value::String(ctx.source_stream.clone()),
);
if let Some(session_id) = &ctx.session_id {
attrs.insert("session_id".to_string(), Value::String(session_id.clone()));
}
attrs
}
pub(crate) fn input_variables_with_context_defaults(
input_variables: &HashMap<String, Value>,
ctx: &SpanEvalContext,
) -> HashMap<String, Value> {
let mut variables = input_mapping_attributes(ctx);
for (key, value) in input_variables {
if !matches!(value, Value::String(s) if s.is_empty()) {
variables.insert(key.clone(), value.clone());
}
}
variables
}
/// Execute scorers on a batch of span records.
///
/// This is the main entry point called from the LlmEvaluation pipeline node handler.
pub async fn execute_scorers(
ctx: &SpanEvalContext,
scorer_refs: &[ScorerRef],
) -> Result<ExecutorOutput> {
let mut output = ExecutorOutput::empty();
let job_input_mapping = match ctx.job_id.as_deref() {
Some(job_id) => infra::table::online_eval_jobs::get_by_org(job_id, &ctx.org_id)
.await?
.and_then(|job| job.input_mapping),
None => None,
};
for scorer_ref in scorer_refs {
let mut resolved_scorer_type = None;
let scorer_lookup = match scorer_ref.version {
Some(v) => {
infra::table::scorers::get_by_entity_id_and_version(&ctx.org_id, &scorer_ref.id, v)
.await
}
None => infra::table::scorers::get_by_entity_id(&ctx.org_id, &scorer_ref.id).await,
};
let score_result = match scorer_lookup {
Ok(Some(scorer)) => {
resolved_scorer_type = Some(scorer.scorer_type);
let attrs = input_mapping_attributes(ctx);
let input_variables =
resolve_scorer_input_variables(&job_input_mapping, &scorer_ref.id, &attrs);
PreparedScorer::from_scorer(&ctx.org_id, scorer)
.await?
.execute(ctx, &input_variables)
.await
}
Ok(None) => {
log::warn!(
"[EXECUTOR-RUNTIME] Scorer entity '{}' not found in org '{}'",
scorer_ref.id,
ctx.org_id
);
Ok(None)
}
Err(e) => Err(anyhow::anyhow!(
"Failed to look up scorer entity '{}': {e}",
scorer_ref.id
)),
};
match score_result {
Ok(Some(score_result)) => {
output.scores.push(score_result.score_json);
output.evaluator_traces.push(score_result.evaluator_trace);
}
Ok(None) => {
log::debug!(
"[EXECUTOR-RUNTIME] Scorer entity '{}' resolved but returned no result",
scorer_ref.id
);
}
Err(e) => {
log::error!(
"[EXECUTOR-RUNTIME] Error executing scorer entity '{}': {e}",
scorer_ref.id
);
let execution_error = e.downcast_ref::<ScorerExecutionError>();
let err_trace = create_evaluator_trace(EvaluatorTraceInput {
org_id: ctx.org_id.clone(),
evaluator_trace_id: ctx.evaluator_trace_id.clone(),
target_span_id: ctx.span_id.clone(),
target_trace_id: ctx.trace_id.clone(),
target_stream: ctx.source_stream.clone(),
scorer_id: Some(scorer_ref.id.clone()),
scorer_version: Some("?".to_string()),
scorer_type: resolved_scorer_type,
job_id: ctx.job_id.clone(),
score_config_id: None,
score_config_version: None,
eval_run_id: ctx.eval_run_id.clone(),
provider_id: execution_error.and_then(|e| e.provider_id.clone()),
provider_name: execution_error.and_then(|e| e.provider_name.clone()),
provider_type: execution_error.and_then(|e| e.provider_type.clone()),
model: execution_error.and_then(|e| e.model.clone()),
latency_ms: execution_error.map(|e| e.latency_ms).unwrap_or(0),
prompt_tokens: execution_error.and_then(|e| e.prompt_tokens),
completion_tokens: execution_error.and_then(|e| e.completion_tokens),
total_tokens: execution_error.and_then(|e| e.total_tokens),
sampling_rate: ctx.sampling_rate,
sampled: ctx.sampled,
status: evaluator::status::ERROR.to_string(),
error_kind: Some("execution_error".to_string()),
error_message: Some(format!("{e}")),
skip_reason: None,
prompt: execution_error.and_then(|e| e.scorer_input.clone()),
response: execution_error.and_then(|e| e.raw_response.clone()),
});
output.evaluator_traces.push(err_trace);
output.errors.push(EvalError {
scorer_id: scorer_ref.id.clone(),
scorer_type: resolved_scorer_type,
error_kind: "execution_error".to_string(),
error_message: format!("{e}"),
latency_ms: execution_error.map(|e| e.latency_ms).unwrap_or(0),
});
}
}
}
Ok(output)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_extract_context_from_basic_span() {
let record = serde_json::json!({
"span_id": "span-1",
"trace_id": "trace-1",
"span_name": "chat completion",
"attributes": {
"gen_ai.system": "openai",
"gen_ai.request.model": "gpt-4o"
}
});
let ctx = extract_context_from_span("org1", Some("job-1"), &record).unwrap();
assert_eq!(ctx.org_id, "org1");
assert_eq!(ctx.span_id, "span-1");
assert_eq!(ctx.trace_id, "trace-1");
assert_eq!(ctx.evaluator_trace_id.len(), 32);
assert_ne!(ctx.evaluator_trace_id, ctx.trace_id);
assert_eq!(ctx.job_id, Some("job-1".to_string()));
assert_eq!(ctx.attributes.get("span_name").unwrap(), "chat completion");
assert_eq!(ctx.attributes.get("gen_ai.system").unwrap(), "openai");
}
#[test]
fn test_extract_context_no_span_id() {
let record = serde_json::json!({
"trace_id": "trace-1",
"some_field": "value"
});
assert!(extract_context_from_span("org1", None, &record).is_none());
}
#[test]
fn test_extract_context_with_session() {
let record = serde_json::json!({
"span_id": "span-2",
"trace_id": "trace-2",
"session_id": "session-1",
"attributes": {}
});
let ctx = extract_context_from_span("org2", None, &record).unwrap();
assert_eq!(ctx.session_id, Some("session-1".to_string()));
assert_eq!(ctx.attributes.is_empty(), true);
}
#[test]
fn test_resolve_scorer_input_variables_from_job_mapping() {
let mapping = Some(
serde_json::from_value(serde_json::json!({
"scorer-1": {
"input": "{{gen_ai.prompt}}",
"output": "{{gen_ai.completion}}"
}
}))
.unwrap(),
);
let attrs = HashMap::from([
("gen_ai.prompt".to_string(), serde_json::json!("question")),
("gen_ai.completion".to_string(), serde_json::json!("answer")),
]);
let variables = resolve_scorer_input_variables(&mapping, "scorer-1", &attrs);
assert_eq!(variables.get("input"), Some(&serde_json::json!("question")));
assert_eq!(variables.get("output"), Some(&serde_json::json!("answer")));
}
#[test]
fn test_input_mapping_attributes_include_span_context_fields() {
let ctx = SpanEvalContext {
org_id: "org1".to_string(),
span_id: "span-1".to_string(),
trace_id: "trace-1".to_string(),
evaluator_trace_id: "11111111111111111111111111111111".to_string(),
session_id: Some("session-1".to_string()),
source_stream: "traces".to_string(),
job_id: Some("job-1".to_string()),
eval_run_id: None,
sampling_rate: None,
sampled: None,
attributes: HashMap::from([("input".to_string(), serde_json::json!("question"))]),
};
let mapping = Some(
serde_json::from_value(serde_json::json!({
"scorer-1": {
"trace_id": "{{trace_id}}",
"span_id": "{{span_id}}",
"session_id": "{{session_id}}",
"stream": "{{pipeline_source_stream}}",
"input": "{{input}}"
}
}))
.unwrap(),
);
let attrs = input_mapping_attributes(&ctx);
let variables = resolve_scorer_input_variables(&mapping, "scorer-1", &attrs);
assert_eq!(
variables.get("trace_id"),
Some(&serde_json::json!("trace-1"))
);
assert_eq!(variables.get("span_id"), Some(&serde_json::json!("span-1")));
assert_eq!(
variables.get("session_id"),
Some(&serde_json::json!("session-1"))
);
assert_eq!(variables.get("stream"), Some(&serde_json::json!("traces")));
assert_eq!(variables.get("input"), Some(&serde_json::json!("question")));
}
#[test]
fn test_input_variables_with_context_defaults_override_empty_context_fields() {
let ctx = SpanEvalContext {
org_id: "org1".to_string(),
span_id: "span-1".to_string(),
trace_id: "trace-1".to_string(),
evaluator_trace_id: "11111111111111111111111111111111".to_string(),
session_id: Some("session-1".to_string()),
source_stream: "traces".to_string(),
job_id: Some("job-1".to_string()),
eval_run_id: None,
sampling_rate: None,
sampled: None,
attributes: HashMap::new(),
};
let input_variables = HashMap::from([
("trace_id".to_string(), serde_json::json!("")),
("input".to_string(), serde_json::json!("question")),
]);
let variables = input_variables_with_context_defaults(&input_variables, &ctx);
assert_eq!(
variables.get("trace_id"),
Some(&serde_json::json!("trace-1"))
);
assert_eq!(variables.get("span_id"), Some(&serde_json::json!("span-1")));
assert_eq!(
variables.get("session_id"),
Some(&serde_json::json!("session-1"))
);
assert_eq!(variables.get("input"), Some(&serde_json::json!("question")));
}
}
+1 -1
View File
@@ -29,8 +29,8 @@ use crate::common::{
utils::auth::{is_ofga_object_visible, remove_ownership, set_ownership},
};
pub mod executor_runtime;
pub mod reconciler;
pub use o2_enterprise::enterprise::llm_evaluations::eval_jobs::executor_runtime;
/// Errors that can occur when interacting with online eval jobs.
#[derive(Debug, thiserror::Error)]
@@ -1,873 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Evaluator Trace — builds internal OTLP spans for online evaluation activity.
//!
//! Evaluator spans are exported to the org's `_evaluator` traces stream through
//! normal OTLP ingestion so the LLM processor can classify LLM-Judge calls and
//! calculate token/cost fields. The evaluator trace ID remains separate from the
//! target application trace; target trace/span identity is stored as evaluator
//! metadata.
use chrono::Utc;
use config::{ider, meta::self_reporting::evaluator};
use infra::table::scorers::ScorerType;
use opentelemetry_proto::tonic::{
collector::trace::v1::ExportTraceServiceRequest,
common::v1::{AnyValue, InstrumentationScope, KeyValue, any_value},
resource::v1::Resource,
trace::v1::{ResourceSpans, ScopeSpans, Span, Status, span::SpanKind, status::StatusCode},
};
use serde::Serialize;
const SERVICE_NAME: &str = "online_eval";
const SCOPE_NAME: &str = "openobserve.online_eval";
const GEN_AI_CHAT_OPERATION: &str = "chat";
#[derive(Debug, Clone, Serialize)]
pub struct EvaluatorTraceInput {
pub org_id: String,
pub evaluator_trace_id: String,
pub target_span_id: String,
pub target_trace_id: String,
pub target_stream: String,
pub scorer_id: Option<String>,
pub scorer_version: Option<String>,
pub scorer_type: Option<ScorerType>,
pub job_id: Option<String>,
pub score_config_id: Option<String>,
pub score_config_version: Option<String>,
pub eval_run_id: Option<String>,
pub provider_id: Option<String>,
pub provider_name: Option<String>,
pub provider_type: Option<String>,
pub model: Option<String>,
pub latency_ms: i64,
pub prompt_tokens: Option<i64>,
pub completion_tokens: Option<i64>,
pub total_tokens: Option<i64>,
pub sampling_rate: Option<f64>,
pub sampled: Option<bool>,
pub status: String,
pub error_kind: Option<String>,
pub error_message: Option<String>,
pub skip_reason: Option<String>,
pub prompt: Option<String>,
pub response: Option<String>,
}
#[derive(Debug, Clone)]
pub struct EvaluatorTrace {
pub span: Span,
pub is_error: bool,
}
pub fn create_evaluator_trace(input: EvaluatorTraceInput) -> EvaluatorTrace {
let span_id = ider::generate_span_id();
let trace_id = input.evaluator_trace_id.clone();
let latency_ms = input.latency_ms.max(0) as u64;
let latency_nanos = latency_ms.saturating_mul(1_000_000);
let end_time = now_nanos();
let start_time = end_time.saturating_sub(latency_nanos);
let is_error =
input.status == evaluator::status::ERROR || input.status == evaluator::status::TIMEOUT;
let is_llm_judge = matches!(input.scorer_type.as_ref(), Some(ScorerType::LlmJudge));
let span_name = if input.status == evaluator::status::SKIPPED {
evaluator::SPAN_NAME_SPAN_SKIPPED
} else {
match input.scorer_type.as_ref() {
Some(ScorerType::LlmJudge) => evaluator::SPAN_NAME_LLM_JUDGE,
Some(ScorerType::Remote) => evaluator::SPAN_NAME_REMOTE_SCORER,
None => evaluator::SPAN_NAME_UNKNOWN_SCORER,
}
};
let mut attributes = Vec::new();
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_TARGET_SPAN_ID),
&input.target_span_id,
);
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_TARGET_TRACE_ID),
&input.target_trace_id,
);
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_TARGET_STREAM),
&input.target_stream,
);
if let Some(ref scorer_id) = input.scorer_id {
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SCORER_ID),
scorer_id,
);
}
if let Some(ref scorer_version) = input.scorer_version {
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SCORER_VERSION),
scorer_version,
);
}
if let Some(ref scorer_type) = input.scorer_type {
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SCORER_TYPE),
scorer_type.to_string(),
);
}
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_JOB_ID),
input.job_id.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SCORE_CONFIG_ID),
input.score_config_id.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SCORE_CONFIG_VERSION),
input.score_config_version.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_EVAL_RUN_ID),
input.eval_run_id.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_PROVIDER_ID),
input.provider_id.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_PROVIDER_NAME),
input.provider_name.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_PROVIDER_TYPE),
input.provider_type.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_MODEL),
input.model.as_deref(),
);
push_i64_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_LATENCY_MS),
input.latency_ms,
);
push_optional_i64_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_PROMPT_TOKENS),
input.prompt_tokens,
);
push_optional_i64_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_COMPLETION_TOKENS),
input.completion_tokens,
);
push_optional_i64_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_TOTAL_TOKENS),
input.total_tokens,
);
if let Some(sampling_rate) = input.sampling_rate {
push_f64_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SAMPLING_RATE),
sampling_rate,
);
}
if let Some(sampled) = input.sampled {
push_bool_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SAMPLED),
sampled,
);
}
push_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_STATUS),
&input.status,
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_ERROR_KIND),
input.error_kind.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_ERROR_MESSAGE),
input.error_message.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_SKIP_REASON),
input.skip_reason.as_deref(),
);
let prompt = input.prompt.as_ref().map(|p| truncate_str(p, 4096));
let response = input.response.as_ref().map(|r| truncate_str(r, 4096));
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_PROMPT),
prompt.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator_attr_key(evaluator::ATTR_RESPONSE),
response.as_deref(),
);
if is_llm_judge {
push_string_attr(
&mut attributes,
evaluator::GEN_AI_OPERATION_NAME,
GEN_AI_CHAT_OPERATION,
);
push_optional_string_attr(
&mut attributes,
evaluator::GEN_AI_REQUEST_MODEL,
input.model.as_deref(),
);
push_optional_string_attr(
&mut attributes,
evaluator::GEN_AI_RESPONSE_MODEL,
input.model.as_deref(),
);
let provider_for_gen_ai = input
.provider_type
.as_deref()
.or(input.provider_name.as_deref());
push_optional_string_attr(
&mut attributes,
evaluator::GEN_AI_PROVIDER_NAME,
provider_for_gen_ai,
);
push_optional_string_attr(
&mut attributes,
evaluator::GEN_AI_SYSTEM,
input.provider_type.as_deref(),
);
if let Some(ref prompt) = prompt {
push_string_attr(
&mut attributes,
evaluator::GEN_AI_INPUT_MESSAGES,
messages_json(prompt, "user"),
);
}
if let Some(ref response) = response {
push_string_attr(
&mut attributes,
evaluator::GEN_AI_OUTPUT_MESSAGES,
messages_json(response, "assistant"),
);
}
push_optional_i64_attr(
&mut attributes,
evaluator::GEN_AI_USAGE_INPUT_TOKENS,
input.prompt_tokens,
);
push_optional_i64_attr(
&mut attributes,
evaluator::GEN_AI_USAGE_OUTPUT_TOKENS,
input.completion_tokens,
);
push_optional_i64_attr(
&mut attributes,
evaluator::GEN_AI_USAGE_TOTAL_TOKENS,
input.total_tokens,
);
}
let span = Span {
trace_id: decode_trace_id(&trace_id),
span_id: decode_span_id(&span_id),
trace_state: String::new(),
parent_span_id: vec![],
flags: 1,
name: span_name.to_string(),
kind: SpanKind::Internal as i32,
start_time_unix_nano: start_time,
end_time_unix_nano: end_time,
attributes,
dropped_attributes_count: 0,
events: vec![],
dropped_events_count: 0,
links: vec![],
dropped_links_count: 0,
status: Some(Status {
code: if is_error {
StatusCode::Error as i32
} else {
StatusCode::Ok as i32
},
message: input.error_message.unwrap_or_default(),
}),
};
EvaluatorTrace { span, is_error }
}
pub fn create_evaluator_trace_request(traces: Vec<EvaluatorTrace>) -> ExportTraceServiceRequest {
let spans: Vec<Span> = traces.into_iter().map(|trace| trace.span).collect();
if spans.is_empty() {
return ExportTraceServiceRequest {
resource_spans: vec![],
};
}
ExportTraceServiceRequest {
resource_spans: vec![ResourceSpans {
resource: Some(Resource {
attributes: vec![string_key_value("service.name", SERVICE_NAME)],
..Default::default()
}),
scope_spans: vec![ScopeSpans {
scope: Some(InstrumentationScope {
name: SCOPE_NAME.to_string(),
version: config::VERSION.to_string(),
attributes: vec![],
dropped_attributes_count: 0,
}),
spans,
..Default::default()
}],
..Default::default()
}],
}
}
fn now_nanos() -> u64 {
Utc::now()
.timestamp_nanos_opt()
.unwrap_or_else(|| Utc::now().timestamp_micros().saturating_mul(1000))
.max(0) as u64
}
fn evaluator_attr_key(key: &str) -> String {
format!("attributes.{key}")
}
fn string_key_value(key: impl Into<String>, value: impl Into<String>) -> KeyValue {
KeyValue {
key: key.into(),
value: Some(AnyValue {
value: Some(any_value::Value::StringValue(value.into())),
}),
..Default::default()
}
}
fn push_string_attr(attrs: &mut Vec<KeyValue>, key: impl Into<String>, value: impl Into<String>) {
attrs.push(string_key_value(key, value));
}
fn push_optional_string_attr(
attrs: &mut Vec<KeyValue>,
key: impl Into<String>,
value: Option<&str>,
) {
if let Some(value) = value {
push_string_attr(attrs, key, value);
}
}
fn push_i64_attr(attrs: &mut Vec<KeyValue>, key: impl Into<String>, value: i64) {
attrs.push(KeyValue {
key: key.into(),
value: Some(AnyValue {
value: Some(any_value::Value::IntValue(value)),
}),
..Default::default()
});
}
fn push_optional_i64_attr(attrs: &mut Vec<KeyValue>, key: impl Into<String>, value: Option<i64>) {
if let Some(value) = value {
push_i64_attr(attrs, key, value);
}
}
fn push_f64_attr(attrs: &mut Vec<KeyValue>, key: impl Into<String>, value: f64) {
attrs.push(KeyValue {
key: key.into(),
value: Some(AnyValue {
value: Some(any_value::Value::DoubleValue(value)),
}),
..Default::default()
});
}
fn push_bool_attr(attrs: &mut Vec<KeyValue>, key: impl Into<String>, value: bool) {
attrs.push(KeyValue {
key: key.into(),
value: Some(AnyValue {
value: Some(any_value::Value::BoolValue(value)),
}),
..Default::default()
});
}
fn decode_trace_id(trace_id: &str) -> Vec<u8> {
decode_hex_id(trace_id, 16).unwrap_or_else(|| {
decode_hex_id(&ider::generate_trace_id(), 16).unwrap_or_else(|| vec![1; 16])
})
}
fn decode_span_id(span_id: &str) -> Vec<u8> {
decode_hex_id(span_id, 8).unwrap_or_else(|| {
decode_hex_id(&ider::generate_span_id(), 8).unwrap_or_else(|| vec![1; 8])
})
}
fn decode_hex_id(id: &str, expected_len: usize) -> Option<Vec<u8>> {
let bytes = hex::decode(id).ok()?;
if bytes.len() == expected_len && bytes.iter().any(|byte| *byte != 0) {
Some(bytes)
} else {
None
}
}
fn messages_json(content: &str, role: &str) -> String {
let trimmed = content.trim();
if let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed)
&& value.is_array()
{
return trimmed.to_string();
}
serde_json::json!([{ "role": role, "content": content }]).to_string()
}
fn truncate_str(s: &str, max_bytes: usize) -> String {
if s.len() <= max_bytes {
return s.to_string();
}
// Walk back to the nearest char boundary so we never slice mid-codepoint.
let mut boundary = max_bytes;
while boundary > 0 && !s.is_char_boundary(boundary) {
boundary -= 1;
}
format!("{}...[truncated]", &s[..boundary])
}
#[cfg(test)]
mod tests {
use super::*;
fn attr<'a>(span: &'a Span, key: &str) -> Option<&'a AnyValue> {
span.attributes
.iter()
.find(|attr| attr.key == key)
.and_then(|attr| attr.value.as_ref())
}
fn attr_str<'a>(span: &'a Span, key: &str) -> Option<&'a str> {
match attr(span, key)?.value.as_ref()? {
any_value::Value::StringValue(value) => Some(value.as_str()),
_ => None,
}
}
fn attr_i64(span: &Span, key: &str) -> Option<i64> {
match attr(span, key)?.value.as_ref()? {
any_value::Value::IntValue(value) => Some(*value),
_ => None,
}
}
fn attr_bool(span: &Span, key: &str) -> Option<bool> {
match attr(span, key)?.value.as_ref()? {
any_value::Value::BoolValue(value) => Some(*value),
_ => None,
}
}
fn attr_f64(span: &Span, key: &str) -> Option<f64> {
match attr(span, key)?.value.as_ref()? {
any_value::Value::DoubleValue(value) => Some(*value),
_ => None,
}
}
fn success_input() -> EvaluatorTraceInput {
EvaluatorTraceInput {
org_id: "org1".to_string(),
evaluator_trace_id: "11111111111111111111111111111111".to_string(),
target_span_id: "span-1".to_string(),
target_trace_id: "trace-1".to_string(),
target_stream: "traces".to_string(),
scorer_id: Some("sc-1".to_string()),
scorer_version: Some("1".to_string()),
scorer_type: Some(ScorerType::LlmJudge),
job_id: Some("job-1".to_string()),
score_config_id: Some("cfg-1".to_string()),
score_config_version: Some("1".to_string()),
eval_run_id: Some("run-1".to_string()),
provider_id: Some("prov-1".to_string()),
provider_name: Some("OpenAI Prod".to_string()),
provider_type: Some("openai".to_string()),
model: Some("gpt-4o".to_string()),
latency_ms: 500,
prompt_tokens: Some(100),
completion_tokens: Some(50),
total_tokens: Some(150),
sampling_rate: Some(0.25),
sampled: Some(true),
status: evaluator::status::SUCCESS.to_string(),
error_kind: None,
error_message: None,
skip_reason: None,
prompt: Some("Evaluate this text...".to_string()),
response: Some("{\"score\":0.9}".to_string()),
}
}
#[test]
fn test_create_success_trace_as_otlp_span() {
let trace = create_evaluator_trace(success_input());
assert!(!trace.is_error);
let span = &trace.span;
assert_eq!(span.name, evaluator::SPAN_NAME_LLM_JUDGE);
assert_eq!(span.kind, SpanKind::Internal as i32);
assert_eq!(span.status.as_ref().unwrap().code, StatusCode::Ok as i32);
assert_eq!(
hex::encode(&span.trace_id),
"11111111111111111111111111111111"
);
assert_eq!(span.span_id.len(), 8);
assert_eq!(span.parent_span_id.len(), 0);
assert_eq!(
span.end_time_unix_nano - span.start_time_unix_nano,
500_000_000
);
assert_eq!(
attr_str(span, &evaluator_attr_key(evaluator::ATTR_TARGET_SPAN_ID)),
Some("span-1")
);
assert_eq!(
attr_str(span, &evaluator_attr_key(evaluator::ATTR_SCORER_ID)),
Some("sc-1")
);
assert_eq!(
attr_i64(span, &evaluator_attr_key(evaluator::ATTR_LATENCY_MS)),
Some(500)
);
assert_eq!(
attr_str(span, &evaluator_attr_key(evaluator::ATTR_PROVIDER_NAME)),
Some("OpenAI Prod")
);
assert_eq!(
attr_str(span, &evaluator_attr_key(evaluator::ATTR_MODEL)),
Some("gpt-4o")
);
assert_eq!(
attr_str(span, &evaluator_attr_key(evaluator::ATTR_STATUS)),
Some(evaluator::status::SUCCESS)
);
assert_eq!(
attr_f64(span, &evaluator_attr_key(evaluator::ATTR_SAMPLING_RATE)),
Some(0.25)
);
assert_eq!(
attr_bool(span, &evaluator_attr_key(evaluator::ATTR_SAMPLED)),
Some(true)
);
assert_eq!(
attr_str(span, evaluator::GEN_AI_OPERATION_NAME),
Some(GEN_AI_CHAT_OPERATION)
);
assert_eq!(
attr_str(span, evaluator::GEN_AI_REQUEST_MODEL),
Some("gpt-4o")
);
assert_eq!(
attr_str(span, evaluator::GEN_AI_RESPONSE_MODEL),
Some("gpt-4o")
);
assert_eq!(
attr_str(span, evaluator::GEN_AI_PROVIDER_NAME),
Some("openai")
);
assert_eq!(
attr_i64(span, evaluator::GEN_AI_USAGE_INPUT_TOKENS),
Some(100)
);
assert_eq!(
attr_i64(span, evaluator::GEN_AI_USAGE_OUTPUT_TOKENS),
Some(50)
);
assert_eq!(
attr_i64(span, evaluator::GEN_AI_USAGE_TOTAL_TOKENS),
Some(150)
);
assert!(
attr_str(span, evaluator::GEN_AI_INPUT_MESSAGES)
.unwrap()
.contains("Evaluate this text")
);
assert!(
attr_str(span, evaluator::GEN_AI_OUTPUT_MESSAGES)
.unwrap()
.contains("score")
);
}
#[test]
fn test_create_error_trace() {
let input = EvaluatorTraceInput {
org_id: "org1".to_string(),
evaluator_trace_id: "22222222222222222222222222222222".to_string(),
target_span_id: "span-2".to_string(),
target_trace_id: "trace-2".to_string(),
target_stream: "traces".to_string(),
scorer_id: Some("sc-2".to_string()),
scorer_version: Some("2".to_string()),
scorer_type: Some(ScorerType::LlmJudge),
job_id: None,
score_config_id: None,
score_config_version: None,
eval_run_id: None,
provider_id: None,
provider_name: None,
provider_type: None,
model: None,
latency_ms: 1000,
prompt_tokens: None,
completion_tokens: None,
total_tokens: None,
sampling_rate: None,
sampled: None,
status: evaluator::status::ERROR.to_string(),
error_kind: Some("timeout".to_string()),
error_message: Some("Request timed out".to_string()),
skip_reason: None,
prompt: None,
response: None,
};
let trace = create_evaluator_trace(input);
assert!(trace.is_error);
assert_eq!(
trace.span.status.as_ref().unwrap().code,
StatusCode::Error as i32
);
assert_eq!(
attr_str(&trace.span, &evaluator_attr_key(evaluator::ATTR_STATUS)),
Some(evaluator::status::ERROR)
);
assert_eq!(
attr_str(&trace.span, &evaluator_attr_key(evaluator::ATTR_ERROR_KIND)),
Some("timeout")
);
assert_eq!(
attr_str(
&trace.span,
&evaluator_attr_key(evaluator::ATTR_ERROR_MESSAGE)
),
Some("Request timed out")
);
}
#[test]
fn test_create_remote_scorer_trace_has_no_gen_ai_attrs() {
let input = EvaluatorTraceInput {
org_id: "org1".to_string(),
evaluator_trace_id: "33333333333333333333333333333333".to_string(),
target_span_id: "span-3".to_string(),
target_trace_id: "trace-3".to_string(),
target_stream: "traces".to_string(),
scorer_id: Some("sc-3".to_string()),
scorer_version: Some("3".to_string()),
scorer_type: Some(ScorerType::Remote),
job_id: None,
score_config_id: None,
score_config_version: None,
eval_run_id: None,
provider_id: None,
provider_name: None,
provider_type: None,
model: None,
latency_ms: 200,
prompt_tokens: None,
completion_tokens: None,
total_tokens: None,
sampling_rate: None,
sampled: None,
status: evaluator::status::SUCCESS.to_string(),
error_kind: None,
error_message: None,
skip_reason: None,
prompt: None,
response: Some("{\"score\":0.9,\"reason\":\"good\"}".to_string()),
};
let trace = create_evaluator_trace(input);
assert!(!trace.is_error);
assert_eq!(trace.span.name, evaluator::SPAN_NAME_REMOTE_SCORER);
assert_eq!(
trace.span.status.as_ref().unwrap().code,
StatusCode::Ok as i32
);
assert_eq!(
attr_str(&trace.span, evaluator::GEN_AI_OPERATION_NAME),
None
);
assert_eq!(
attr_str(&trace.span, &evaluator_attr_key(evaluator::ATTR_RESPONSE)),
Some("{\"score\":0.9,\"reason\":\"good\"}")
);
}
#[test]
fn test_create_skipped_trace() {
let input = EvaluatorTraceInput {
org_id: "org1".to_string(),
evaluator_trace_id: "55555555555555555555555555555555".to_string(),
target_span_id: "span-5".to_string(),
target_trace_id: "trace-5".to_string(),
target_stream: "traces".to_string(),
scorer_id: None,
scorer_version: None,
scorer_type: None,
job_id: Some("job-5".to_string()),
score_config_id: None,
score_config_version: None,
eval_run_id: Some("run-5".to_string()),
provider_id: None,
provider_name: None,
provider_type: None,
model: None,
latency_ms: 0,
prompt_tokens: None,
completion_tokens: None,
total_tokens: None,
sampling_rate: Some(0.1),
sampled: Some(false),
status: evaluator::status::SKIPPED.to_string(),
error_kind: None,
error_message: None,
skip_reason: Some("sampling".to_string()),
prompt: None,
response: None,
};
let trace = create_evaluator_trace(input);
assert!(!trace.is_error);
assert_eq!(trace.span.name, evaluator::SPAN_NAME_SPAN_SKIPPED);
assert_eq!(
trace.span.status.as_ref().unwrap().code,
StatusCode::Ok as i32
);
assert_eq!(
attr_str(&trace.span, &evaluator_attr_key(evaluator::ATTR_STATUS)),
Some(evaluator::status::SKIPPED)
);
assert_eq!(
attr_str(
&trace.span,
&evaluator_attr_key(evaluator::ATTR_SKIP_REASON)
),
Some("sampling")
);
assert_eq!(
attr_bool(&trace.span, &evaluator_attr_key(evaluator::ATTR_SAMPLED)),
Some(false)
);
assert_eq!(
attr_str(&trace.span, &evaluator_attr_key(evaluator::ATTR_SCORER_ID)),
None
);
}
#[test]
fn test_multiple_scorer_spans_share_evaluator_trace_id() {
let evaluator_trace_id = "11111111111111111111111111111111".to_string();
let make_input = |scorer_id: &str| EvaluatorTraceInput {
evaluator_trace_id: evaluator_trace_id.clone(),
scorer_id: Some(scorer_id.to_string()),
..success_input()
};
let first = create_evaluator_trace(make_input("sc-1"));
let second = create_evaluator_trace(make_input("sc-2"));
assert_eq!(hex::encode(&first.span.trace_id), evaluator_trace_id);
assert_eq!(first.span.trace_id, second.span.trace_id);
assert_ne!(first.span.span_id, second.span.span_id);
}
#[test]
fn test_create_evaluator_trace_request_groups_spans_under_resource() {
let trace = create_evaluator_trace(success_input());
let request = create_evaluator_trace_request(vec![trace]);
assert_eq!(request.resource_spans.len(), 1);
let resource_spans = &request.resource_spans[0];
assert_eq!(
resource_spans.resource.as_ref().unwrap().attributes[0].key,
"service.name"
);
assert_eq!(resource_spans.scope_spans.len(), 1);
assert_eq!(
resource_spans.scope_spans[0].scope.as_ref().unwrap().name,
SCOPE_NAME
);
assert_eq!(resource_spans.scope_spans[0].spans.len(), 1);
}
#[test]
fn test_messages_json_wraps_plain_content_and_preserves_arrays() {
assert_eq!(
messages_json("hello", "user"),
serde_json::json!([{ "role": "user", "content": "hello" }]).to_string()
);
let messages = r#"[{"role":"system","content":"rules"}]"#;
assert_eq!(messages_json(messages, "user"), messages);
}
#[test]
fn test_truncate_str() {
assert_eq!(truncate_str("hello", 10), "hello");
assert_eq!(truncate_str("hello world", 5), "hello...[truncated]");
}
}
+4 -5
View File
@@ -14,9 +14,8 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
pub mod eval_jobs;
pub mod evaluator_trace;
pub mod evaluator_trace_exporter;
pub mod prepared_scorers;
pub mod score_configs;
pub mod score_writer;
pub mod scorers;
pub use o2_enterprise::enterprise::llm_evaluations::{
evaluator_trace, prepared_scorers, score_configs, score_writer, scorers,
};
@@ -1,485 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{collections::HashMap, fmt};
use anyhow::Result;
use config::meta::self_reporting::llm_scores::{LlmScoreDataSourceType, LlmScoreDataType};
use infra::table::{score_configs::ScoreConfigDataType, scorers::ScorerType};
use serde_json::Value;
use crate::service::llm_evaluations::{
eval_jobs::executor_runtime::{SpanEvalContext, input_variables_with_context_defaults},
evaluator_trace::{EvaluatorTrace, EvaluatorTraceInput, create_evaluator_trace},
score_writer::{ScoreWriterInput, create_score},
scorers::{llm_judge, remote_client, schema_derivation::ExtraMetadataField},
};
#[derive(Debug)]
pub(crate) struct ScorerExecutionError {
message: String,
pub(crate) raw_response: Option<String>,
pub(crate) scorer_input: Option<String>,
pub(crate) provider_id: Option<String>,
pub(crate) provider_name: Option<String>,
pub(crate) provider_type: Option<String>,
pub(crate) model: Option<String>,
pub(crate) prompt_tokens: Option<i64>,
pub(crate) completion_tokens: Option<i64>,
pub(crate) total_tokens: Option<i64>,
pub(crate) latency_ms: i64,
}
impl fmt::Display for ScorerExecutionError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.message)
}
}
impl std::error::Error for ScorerExecutionError {}
pub(crate) struct ScoreResult {
pub(crate) score_json: Value,
pub(crate) evaluator_trace: EvaluatorTrace,
}
pub(crate) enum PreparedScorer {
LlmJudge(PreparedLlmJudgeScorer),
Remote(PreparedRemoteScorer),
}
impl PreparedScorer {
pub(crate) async fn from_scorer(
org_id: &str,
scorer: infra::table::scorers::Scorer,
) -> Result<Self> {
match scorer.scorer_type {
ScorerType::LlmJudge => Ok(Self::LlmJudge(
PreparedLlmJudgeScorer::prepare(org_id, &scorer).await?,
)),
ScorerType::Remote => Ok(Self::Remote(
PreparedRemoteScorer::prepare(org_id, &scorer).await?,
)),
}
}
pub(crate) async fn execute(
&self,
span_ctx: &SpanEvalContext,
input_variables: &HashMap<String, Value>,
) -> Result<Option<ScoreResult>> {
match self {
Self::LlmJudge(scorer) => scorer.execute(span_ctx, input_variables).await,
Self::Remote(scorer) => scorer.execute(span_ctx, input_variables).await,
}
}
}
pub struct PreparedLlmJudgeScorer {
scorer: infra::table::scorers::Scorer,
provider: infra::provider::PreparedProvider,
scorer_cfg: llm_judge::ScorerConfig,
score_cfg_info: llm_judge::ScoreConfigInfo,
extra_metadata_fields: Vec<ExtraMetadataField>,
include_reasoning: bool,
}
impl PreparedLlmJudgeScorer {
pub async fn prepare(org_id: &str, scorer: &infra::table::scorers::Scorer) -> Result<Self> {
Self::prepare_with_score_config_info(org_id, scorer, None).await
}
pub async fn prepare_with_score_config_info(
org_id: &str,
scorer: &infra::table::scorers::Scorer,
score_config_info: Option<llm_judge::ScoreConfigInfo>,
) -> Result<Self> {
use llm_judge::{LlmJudgeParams, ScoreConfigInfo, ScorerConfig};
let params: LlmJudgeParams = serde_json::from_value(scorer.params.clone())
.map_err(|e| anyhow::anyhow!("Invalid LLM judge scorer params: {e}"))?;
let provider_id = params.provider_id.as_str();
let provider = match infra::table::providers::get(provider_id).await {
Ok(Some(p)) => p,
Ok(None) => {
anyhow::bail!("Provider '{provider_id}' not found for LLM judge scorer");
}
Err(e) => {
anyhow::bail!("Failed to look up provider '{provider_id}': {e}");
}
};
let score_config_entity_id = scorer.produces_score_config_id.as_deref().unwrap_or("");
let score_cfg_info = if let Some(score_config_info) = score_config_info {
score_config_info
} else if !score_config_entity_id.is_empty() {
lookup_score_config_for_scorer(org_id, scorer)
.await
.as_ref()
.map(ScoreConfigInfo::from)
.unwrap_or_default()
} else {
ScoreConfigInfo::default()
};
let template = if scorer.template.trim().is_empty() {
"Evaluate the given content and produce a score.".to_string()
} else {
scorer.template.clone()
};
let include_reasoning = params.include_reasoning.unwrap_or(true);
let extra_metadata_fields = params.extra_metadata_fields.clone().unwrap_or_default();
let scorer_cfg = ScorerConfig {
scorer_entity_id: scorer.entity_id.clone(),
scorer_version: scorer.version.to_string(),
scorer_type: scorer.scorer_type,
template,
params,
};
let provider = infra::provider::PreparedProvider::parse((&provider).into())?;
Ok(Self {
scorer: scorer.clone(),
provider,
scorer_cfg,
score_cfg_info,
extra_metadata_fields,
include_reasoning,
})
}
pub async fn run(
&self,
input_variables: &HashMap<String, Value>,
) -> Result<llm_judge::LlmJudgeOutput> {
llm_judge::run_llm_judge(
&self.provider,
&self.scorer_cfg,
&self.score_cfg_info,
input_variables,
&self.extra_metadata_fields,
self.include_reasoning,
)
.await
}
async fn execute(
&self,
span_ctx: &SpanEvalContext,
input_variables: &HashMap<String, Value>,
) -> Result<Option<ScoreResult>> {
let start = std::time::Instant::now();
match self.run(input_variables).await {
Ok(llm_output) => {
let latency = start.elapsed();
let (value_numeric, value_categorical, value_boolean, data_type) =
match self.score_cfg_info.data_type {
ScoreConfigDataType::Numeric => (
llm_output.value_numeric,
None,
None,
LlmScoreDataType::Numeric,
),
ScoreConfigDataType::Categorical => (
None,
llm_output.value_categorical,
None,
LlmScoreDataType::Categorical,
),
ScoreConfigDataType::Boolean => (
None,
None,
llm_output.value_boolean,
LlmScoreDataType::Boolean,
),
};
let score_name = self
.score_cfg_info
.score_config_name
.clone()
.unwrap_or_else(|| self.scorer.name.clone());
let score = create_score(ScoreWriterInput {
span_id: span_ctx.span_id.clone(),
trace_id: span_ctx.trace_id.clone(),
session_id: span_ctx.session_id.clone(),
scorer_id: self.scorer_cfg.scorer_entity_id.clone(),
scorer_version: self.scorer_cfg.scorer_version.clone(),
score_config_id: self.score_cfg_info.score_config_entity_id.clone(),
score_config_version: self.score_cfg_info.score_config_version.clone(),
score_name: Some(score_name),
source_type: LlmScoreDataSourceType::LlmJudge,
source_stream: Some(span_ctx.source_stream.clone()),
job_id: span_ctx.job_id.clone(),
value_numeric,
value_categorical,
value_boolean,
data_type,
reasoning: llm_output.reasoning.clone(),
metadata: llm_output.metadata.clone(),
author: None,
});
let trace = create_evaluator_trace(EvaluatorTraceInput {
org_id: span_ctx.org_id.clone(),
evaluator_trace_id: span_ctx.evaluator_trace_id.clone(),
target_span_id: span_ctx.span_id.clone(),
target_trace_id: span_ctx.trace_id.clone(),
target_stream: span_ctx.source_stream.clone(),
scorer_id: Some(self.scorer_cfg.scorer_entity_id.clone()),
scorer_version: Some(self.scorer_cfg.scorer_version.clone()),
scorer_type: Some(self.scorer_cfg.scorer_type),
job_id: span_ctx.job_id.clone(),
score_config_id: self.score_cfg_info.score_config_entity_id.clone(),
score_config_version: self.score_cfg_info.score_config_version.clone(),
eval_run_id: span_ctx.eval_run_id.clone(),
provider_id: Some(self.provider.id.clone()),
provider_name: Some(self.provider.name.clone()),
provider_type: Some(self.provider.kind.as_str().to_string()),
model: Some(llm_output.model_used),
latency_ms: latency.as_millis() as i64,
prompt_tokens: llm_output.prompt_tokens,
completion_tokens: llm_output.completion_tokens,
total_tokens: llm_output.total_tokens,
sampling_rate: span_ctx.sampling_rate,
sampled: span_ctx.sampled,
status: config::meta::self_reporting::evaluator::status::SUCCESS.to_string(),
error_kind: None,
error_message: None,
skip_reason: None,
prompt: Some(llm_output.prompt_messages),
response: Some(llm_output.raw_response),
});
Ok(Some(ScoreResult {
score_json: score.json,
evaluator_trace: trace,
}))
}
Err(e) => {
let latency = start.elapsed();
let structured_error = e.chain().find_map(|cause| {
cause.downcast_ref::<infra::provider::ProviderStructuredOutputError>()
});
let prompt_tokens = structured_error.and_then(|e| e.prompt_tokens);
let completion_tokens = structured_error.and_then(|e| e.completion_tokens);
let total_tokens = structured_error
.and_then(|e| e.total_tokens)
.or_else(|| prompt_tokens.zip(completion_tokens).map(|(p, c)| p + c));
Err(ScorerExecutionError {
message: format!(
"LLM Judge scorer failed: {e} (latency: {}ms)",
latency.as_millis()
),
raw_response: structured_error.map(|e| e.raw_response.clone()),
scorer_input: Some(llm_judge::build_prompt_messages_json(
&self.scorer_cfg,
input_variables,
)),
provider_id: Some(self.provider.id.clone()),
provider_name: Some(self.provider.name.clone()),
provider_type: Some(self.provider.kind.as_str().to_string()),
model: structured_error.map(|e| e.model_used.clone()).or_else(|| {
Some(
self.provider
.model_or_default(self.scorer_cfg.params.model.as_deref()),
)
}),
prompt_tokens,
completion_tokens,
total_tokens,
latency_ms: structured_error
.map(|e| e.latency_ms)
.unwrap_or(latency.as_millis() as i64),
}
.into())
}
}
}
}
pub struct PreparedRemoteScorer {
scorer: infra::table::scorers::Scorer,
remote_cfg: remote_client::RemoteScorerConfig,
score_config: Option<infra::table::score_configs::ScoreConfig>,
}
impl PreparedRemoteScorer {
pub async fn prepare(org_id: &str, scorer: &infra::table::scorers::Scorer) -> Result<Self> {
let mut remote_cfg = remote_client::parse_remote_config(&scorer.params)?;
if !scorer.template.trim().is_empty() {
remote_cfg.request_body_template = Some(scorer.template.clone());
}
let score_config = match scorer.produces_score_config_id.as_deref() {
Some(entity_id) if !entity_id.is_empty() => {
lookup_score_config_for_scorer(org_id, scorer).await
}
_ => None,
};
Ok(Self {
scorer: scorer.clone(),
remote_cfg,
score_config,
})
}
pub async fn run(
&self,
input_variables: &HashMap<String, Value>,
) -> Result<remote_client::RemoteScorerOutput> {
remote_client::run_remote_scorer(&self.remote_cfg, input_variables).await
}
async fn execute(
&self,
span_ctx: &SpanEvalContext,
input_variables: &HashMap<String, Value>,
) -> Result<Option<ScoreResult>> {
let start = std::time::Instant::now();
let input_variables = input_variables_with_context_defaults(input_variables, span_ctx);
match self.run(&input_variables).await {
Ok(remote_output) => {
let latency = start.elapsed();
let value = &remote_output.value;
let (value_numeric, value_categorical, value_boolean) = if value.is_number() {
(value.as_f64(), None, None)
} else if value.is_string() {
(None, value.as_str().map(|s| s.to_string()), None)
} else if value.is_boolean() {
(None, None, value.as_bool())
} else {
(None, Some(value.to_string()), None)
};
let data_type = if value_numeric.is_some() {
LlmScoreDataType::Numeric
} else if value_boolean.is_some() {
LlmScoreDataType::Boolean
} else {
LlmScoreDataType::Categorical
};
let score_name = self
.score_config
.as_ref()
.map(|sc| sc.name.clone())
.unwrap_or_else(|| self.scorer.name.clone());
let score = create_score(ScoreWriterInput {
span_id: span_ctx.span_id.clone(),
trace_id: span_ctx.trace_id.clone(),
session_id: span_ctx.session_id.clone(),
scorer_id: self.scorer.entity_id.clone(),
scorer_version: self.scorer.version.to_string(),
score_config_id: self.score_config.as_ref().map(|sc| sc.entity_id.clone()),
score_config_version: self
.score_config
.as_ref()
.map(|sc| sc.version.to_string()),
score_name: Some(score_name),
source_type: LlmScoreDataSourceType::Remote,
source_stream: Some(span_ctx.source_stream.clone()),
job_id: span_ctx.job_id.clone(),
value_numeric,
value_categorical,
value_boolean,
data_type,
reasoning: remote_output.reasoning.clone(),
metadata: remote_output.metadata.clone(),
author: None,
});
let trace = create_evaluator_trace(EvaluatorTraceInput {
org_id: span_ctx.org_id.clone(),
evaluator_trace_id: span_ctx.evaluator_trace_id.clone(),
target_span_id: span_ctx.span_id.clone(),
target_trace_id: span_ctx.trace_id.clone(),
target_stream: span_ctx.source_stream.clone(),
scorer_id: Some(self.scorer.entity_id.clone()),
scorer_version: Some(self.scorer.version.to_string()),
scorer_type: Some(self.scorer.scorer_type),
job_id: span_ctx.job_id.clone(),
score_config_id: self.score_config.as_ref().map(|sc| sc.entity_id.clone()),
score_config_version: self
.score_config
.as_ref()
.map(|sc| sc.version.to_string()),
eval_run_id: span_ctx.eval_run_id.clone(),
provider_id: None,
provider_name: None,
provider_type: None,
model: None,
latency_ms: latency.as_millis() as i64,
prompt_tokens: None,
completion_tokens: None,
total_tokens: None,
sampling_rate: span_ctx.sampling_rate,
sampled: span_ctx.sampled,
status: config::meta::self_reporting::evaluator::status::SUCCESS.to_string(),
error_kind: None,
error_message: None,
skip_reason: None,
prompt: None,
response: Some(remote_output.raw_response),
});
Ok(Some(ScoreResult {
score_json: score.json,
evaluator_trace: trace,
}))
}
Err(e) => {
let latency = start.elapsed();
anyhow::bail!(
"Remote scorer failed: {e} (latency: {}ms)",
latency.as_millis()
);
}
}
}
}
async fn lookup_score_config_for_scorer(
org_id: &str,
scorer: &infra::table::scorers::Scorer,
) -> Option<infra::table::score_configs::ScoreConfig> {
let entity_id = scorer.produces_score_config_id.as_deref()?;
match scorer.produces_score_config_version {
Some(version) => {
infra::table::score_configs::get_by_entity_id_and_version(org_id, entity_id, version)
.await
.ok()
.flatten()
}
None => infra::table::score_configs::get_by_entity_id(org_id, entity_id)
.await
.ok()
.flatten(),
}
}
@@ -1,250 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use chrono::Utc;
use config::ider;
use infra::table;
use crate::common::{
meta::authz::Authz,
utils::auth::{is_ofga_object_visible, remove_ownership, set_ownership},
};
/// Errors that can occur when interacting with score configs.
#[derive(Debug, thiserror::Error)]
pub enum ScoreConfigError {
#[error("InfraError# Internal error")]
InfraError(#[from] infra::errors::Error),
#[error("Score config name cannot be empty")]
MissingName,
#[error("Score config not found")]
NotFound,
#[error("Score config name already exists")]
DuplicateName,
#[error("Score config cannot be deleted because it is used by one or more scorers")]
InUseByScorer,
}
#[tracing::instrument(skip(config))]
pub async fn save_score_config(
org_id: &str,
mut config: table::score_configs::ScoreConfig,
) -> Result<table::score_configs::ScoreConfig, ScoreConfigError> {
config.name = config.name.trim().to_string();
if config.name.is_empty() {
return Err(ScoreConfigError::MissingName);
}
if table::score_configs::get_by_active_name(org_id, &config.name)
.await?
.is_some()
{
return Err(ScoreConfigError::DuplicateName);
}
config.id = ider::generate();
config.entity_id = ider::generate();
config.org_id = org_id.to_string();
config.version = 1;
let now = Utc::now().timestamp_millis();
config.created_at = now;
config.updated_at = now;
table::score_configs::add(&config).await?;
set_ownership(org_id, "score_configs", Authz::new(&config.entity_id)).await;
publish_score_config_put(&config).await;
Ok(config)
}
#[tracing::instrument(skip(config))]
pub async fn update_score_config(
org_id: &str,
entity_id: &str,
mut config: table::score_configs::ScoreConfig,
) -> Result<table::score_configs::ScoreConfig, ScoreConfigError> {
// Get existing latest
let existing = table::score_configs::get_by_entity_id(org_id, entity_id)
.await?
.ok_or(ScoreConfigError::NotFound)?;
let new_name = config.name.trim().to_string();
let new_name = if new_name.is_empty() {
existing.name.clone()
} else {
new_name
};
if new_name != existing.name
&& table::score_configs::get_by_active_name(org_id, &new_name)
.await?
.is_some()
{
return Err(ScoreConfigError::DuplicateName);
}
let latest_version = table::score_configs::get_latest_version(org_id, entity_id).await?;
config.id = ider::generate();
config.entity_id = existing.entity_id;
config.org_id = org_id.to_string();
config.version = latest_version + 1;
let now = Utc::now().timestamp_millis();
config.created_at = now;
config.updated_at = now;
config.is_active = true;
config.name = new_name;
// data_type is immutable - preserve from existing
config.data_type = existing.data_type;
table::score_configs::delete(entity_id, org_id).await?;
table::score_configs::add(&config).await?;
publish_score_config_put(&config).await;
Ok(config)
}
#[tracing::instrument()]
pub async fn list_score_configs(
org_id: &str,
permitted_objects: Option<Vec<String>>,
) -> Result<Vec<table::score_configs::ScoreConfig>, ScoreConfigError> {
let configs = table::score_configs::get_all_by_org(org_id).await?;
Ok(configs
.into_iter()
.filter(|config| {
is_ofga_object_visible(
org_id,
"score_config",
&config.entity_id,
permitted_objects.as_deref(),
)
})
.collect())
}
#[tracing::instrument()]
pub async fn get_score_config(
org_id: &str,
entity_id: &str,
) -> Result<table::score_configs::ScoreConfig, ScoreConfigError> {
let config = table::score_configs::get_by_entity_id(org_id, entity_id)
.await?
.ok_or(ScoreConfigError::NotFound)?;
Ok(config)
}
#[tracing::instrument()]
pub async fn get_score_config_versions(
org_id: &str,
entity_id: &str,
) -> Result<Vec<table::score_configs::ScoreConfig>, ScoreConfigError> {
let versions = table::score_configs::get_versions(org_id, entity_id).await?;
Ok(versions)
}
#[tracing::instrument()]
pub async fn delete_score_config(org_id: &str, entity_id: &str) -> Result<(), ScoreConfigError> {
table::score_configs::get_by_entity_id(org_id, entity_id)
.await?
.ok_or(ScoreConfigError::NotFound)?;
if table::scorers::has_active_by_score_config_id(org_id, entity_id).await? {
return Err(ScoreConfigError::InUseByScorer);
}
table::score_configs::delete(entity_id, org_id).await?;
remove_ownership(org_id, "score_configs", Authz::new(entity_id)).await;
publish_score_config_delete(org_id, entity_id).await;
Ok(())
}
#[cfg(feature = "enterprise")]
async fn publish_score_config_put(config: &table::score_configs::ScoreConfig) {
if !o2_enterprise::enterprise::common::config::get_config()
.super_cluster
.enabled
{
return;
}
let key = format!("/eval/score_configs/{}/{}", config.org_id, config.entity_id);
let message = o2_enterprise::enterprise::super_cluster::queue::EvalScoreConfigMessage::Put {
config: config.clone(),
};
match config::utils::json::to_vec(&message) {
Ok(value) => {
if let Err(e) = o2_enterprise::enterprise::super_cluster::queue::eval_score_config_put(
&key,
value.into(),
)
.await
{
log::error!("[ScoreConfig] error publishing super cluster score config put: {e}");
}
}
Err(e) => {
log::error!("[ScoreConfig] error serializing score config super cluster message: {e}")
}
}
}
#[cfg(not(feature = "enterprise"))]
async fn publish_score_config_put(_config: &table::score_configs::ScoreConfig) {}
#[cfg(feature = "enterprise")]
async fn publish_score_config_delete(org_id: &str, entity_id: &str) {
if !o2_enterprise::enterprise::common::config::get_config()
.super_cluster
.enabled
{
return;
}
let key = format!("/eval/score_configs/{}/{}", org_id, entity_id);
if let Err(e) =
o2_enterprise::enterprise::super_cluster::queue::eval_score_config_delete(&key).await
{
log::error!("[ScoreConfig] error publishing super cluster score config delete: {e}");
}
}
#[cfg(not(feature = "enterprise"))]
async fn publish_score_config_delete(_org_id: &str, _entity_id: &str) {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_score_config_error_variants() {
let infra_error =
ScoreConfigError::InfraError(infra::errors::Error::Message("test".to_string()));
assert!(matches!(infra_error, ScoreConfigError::InfraError(_)));
let missing_name = ScoreConfigError::MissingName;
assert!(matches!(missing_name, ScoreConfigError::MissingName));
let not_found = ScoreConfigError::NotFound;
assert!(matches!(not_found, ScoreConfigError::NotFound));
let duplicate_name = ScoreConfigError::DuplicateName;
assert!(matches!(duplicate_name, ScoreConfigError::DuplicateName));
let in_use_by_scorer = ScoreConfigError::InUseByScorer;
assert!(matches!(in_use_by_scorer, ScoreConfigError::InUseByScorer));
}
}
-242
View File
@@ -1,242 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Score Writer — writes evaluation results to the `_llm_scores` system stream.
//!
//! Produces `LlmScoreRecord` values from scorer execution outputs and
//! serializes them for ingestion through the pipeline's output node.
use chrono::Utc;
use config::{
ider,
meta::self_reporting::llm_scores::{
LlmScoreDataLevel, LlmScoreDataSourceType, LlmScoreDataType, LlmScoreRecord,
},
};
#[derive(Debug)]
pub struct ScoreWriterInput {
pub span_id: String,
pub trace_id: String,
pub session_id: Option<String>,
pub scorer_id: String,
pub scorer_version: String,
pub score_config_id: Option<String>,
pub score_config_version: Option<String>,
pub score_name: Option<String>,
pub source_type: LlmScoreDataSourceType,
pub source_stream: Option<String>,
pub job_id: Option<String>,
pub value_numeric: Option<f64>,
pub value_categorical: Option<String>,
pub value_boolean: Option<bool>,
pub data_type: LlmScoreDataType,
pub reasoning: Option<String>,
pub metadata: Option<serde_json::Value>,
pub author: Option<String>,
}
pub struct WrittenScore {
pub record: LlmScoreRecord,
pub json: serde_json::Value,
}
pub fn create_score(input: ScoreWriterInput) -> WrittenScore {
let now = Utc::now().timestamp_micros();
let id = ider::generate();
let level = if input.span_id.is_empty() {
LlmScoreDataLevel::Trace
} else {
LlmScoreDataLevel::Span
};
let record = LlmScoreRecord {
id,
span_id: if input.span_id.is_empty() {
None
} else {
Some(input.span_id)
},
trace_id: if input.trace_id.is_empty() {
None
} else {
Some(input.trace_id)
},
session_id: input.session_id,
experiment_id: None,
level,
name: input.score_name.unwrap_or_else(|| "unknown".to_string()),
value_numeric: input.value_numeric,
value_categorical: input.value_categorical,
value_boolean: input.value_boolean,
data_type: input.data_type,
scorer_id: Some(input.scorer_id),
scorer_version: Some(input.scorer_version),
score_config_id: input.score_config_id,
score_config_version: input.score_config_version,
source_type: input.source_type,
source_stream: input.source_stream,
job_id: input.job_id,
reasoning: input.reasoning,
metadata: input.metadata,
author: input.author,
_timestamp: now,
};
let json = serde_json::to_value(&record)
.unwrap_or_else(|_| serde_json::json!({"error": "serialization_failed"}));
WrittenScore { record, json }
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_create_numeric_score() {
let input = ScoreWriterInput {
span_id: "span-1".to_string(),
trace_id: "trace-1".to_string(),
session_id: None,
scorer_id: "scorer-entity-1".to_string(),
scorer_version: "1".to_string(),
score_config_id: Some("cfg-entity-1".to_string()),
score_config_version: Some("1".to_string()),
score_name: Some("faithfulness".to_string()),
source_type: LlmScoreDataSourceType::LlmJudge,
source_stream: Some("traces".to_string()),
job_id: Some("job-1".to_string()),
value_numeric: Some(0.95),
value_categorical: None,
value_boolean: None,
data_type: LlmScoreDataType::Numeric,
reasoning: None,
metadata: None,
author: None,
};
let result = create_score(input);
assert_eq!(result.record.level, LlmScoreDataLevel::Span);
assert_eq!(result.record.name, "faithfulness");
assert_eq!(result.record.value_numeric, Some(0.95));
assert_eq!(result.record.source_type, LlmScoreDataSourceType::LlmJudge);
assert_eq!(result.record.scorer_id, Some("scorer-entity-1".to_string()));
assert_eq!(result.record.job_id, Some("job-1".to_string()));
assert_eq!(
result.record.score_config_id,
Some("cfg-entity-1".to_string())
);
let json = &result.json;
assert_eq!(json["name"], "faithfulness");
assert_eq!(json["level"], "span");
assert_eq!(json["scorer_id"], "scorer-entity-1");
assert_eq!(json["source_type"], "llm_judge");
assert_eq!(json["score_config_id"], "cfg-entity-1");
assert!(json.get("score_config_entity_id").is_none());
}
#[test]
fn test_create_categorical_score() {
let input = ScoreWriterInput {
span_id: "span-2".to_string(),
trace_id: "trace-2".to_string(),
session_id: None,
scorer_id: "sc-2".to_string(),
scorer_version: "2".to_string(),
score_config_id: None,
score_config_version: None,
score_name: Some("quality".to_string()),
source_type: LlmScoreDataSourceType::Remote,
source_stream: None,
job_id: None,
value_numeric: None,
value_categorical: Some("excellent".to_string()),
value_boolean: None,
data_type: LlmScoreDataType::Categorical,
reasoning: Some("good job".to_string()),
metadata: None,
author: None,
};
let result = create_score(input);
assert_eq!(result.record.data_type, LlmScoreDataType::Categorical);
assert_eq!(
result.record.value_categorical,
Some("excellent".to_string())
);
assert_eq!(result.record.source_type, LlmScoreDataSourceType::Remote);
assert_eq!(result.record.reasoning, Some("good job".to_string()));
}
#[test]
fn test_create_boolean_score() {
let input = ScoreWriterInput {
span_id: String::new(),
trace_id: "trace-3".to_string(),
session_id: None,
scorer_id: "sc-3".to_string(),
scorer_version: "3".to_string(),
score_config_id: None,
score_config_version: None,
score_name: Some("pass_fail".to_string()),
source_type: LlmScoreDataSourceType::LlmJudge,
source_stream: None,
job_id: None,
value_numeric: None,
value_categorical: None,
value_boolean: Some(true),
data_type: LlmScoreDataType::Boolean,
reasoning: None,
metadata: None,
author: Some("admin".to_string()),
};
let result = create_score(input);
assert_eq!(result.record.level, LlmScoreDataLevel::Trace);
assert_eq!(result.record.value_boolean, Some(true));
assert_eq!(result.record.author, Some("admin".to_string()));
}
#[test]
fn test_create_score_with_metadata() {
let input = ScoreWriterInput {
span_id: "span-4".to_string(),
trace_id: "trace-4".to_string(),
session_id: Some("session-1".to_string()),
scorer_id: "sc-4".to_string(),
scorer_version: "4".to_string(),
score_config_id: None,
score_config_version: None,
score_name: Some("custom".to_string()),
source_type: LlmScoreDataSourceType::LlmJudge,
source_stream: Some("traces".to_string()),
job_id: None,
value_numeric: Some(0.5),
value_categorical: None,
value_boolean: None,
data_type: LlmScoreDataType::Numeric,
reasoning: None,
metadata: Some(serde_json::json!({"extra": "info"})),
author: None,
};
let result = create_score(input);
assert_eq!(result.record.session_id, Some("session-1".to_string()));
assert!(result.record.metadata.is_some());
}
}
@@ -1,287 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! LLM Judge scorer execution.
use std::{collections::HashMap, time::Instant};
use anyhow::Result;
use infra::{
provider::{PreparedProvider, ProviderRunParams},
table::{score_configs::ScoreConfigDataType, scorers::ScorerType},
};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use super::schema_derivation::{ExtraMetadataField, REASONING_FIELD, derive_output_schema};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LlmJudgeOutput {
pub value_numeric: Option<f64>,
pub value_categorical: Option<String>,
pub value_boolean: Option<bool>,
pub reasoning: Option<String>,
pub metadata: Option<Value>,
pub raw_response: String,
pub prompt_messages: String,
pub prompt_tokens: Option<i64>,
pub completion_tokens: Option<i64>,
pub total_tokens: Option<i64>,
pub model_used: String,
pub latency_ms: i64,
}
#[derive(Debug, Clone)]
pub struct ScorerConfig {
pub scorer_entity_id: String,
pub scorer_version: String,
pub scorer_type: ScorerType,
pub template: String,
pub params: LlmJudgeParams,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct LlmJudgeParams {
pub provider_id: String,
pub model: Option<String>,
pub temperature: Option<f64>,
pub max_tokens: Option<u32>,
pub timeout_ms: Option<u64>,
pub output_parsing: Option<String>,
pub include_reasoning: Option<bool>,
pub extra_metadata_fields: Option<Vec<ExtraMetadataField>>,
}
#[derive(Debug, Clone, Default)]
pub struct ScoreConfigInfo {
pub score_config_entity_id: Option<String>,
pub score_config_version: Option<String>,
pub score_config_name: Option<String>,
pub data_type: ScoreConfigDataType,
pub numeric_range: Option<Value>,
pub categories: Option<Value>,
}
impl From<&infra::table::score_configs::ScoreConfig> for ScoreConfigInfo {
fn from(score_config: &infra::table::score_configs::ScoreConfig) -> Self {
Self {
score_config_entity_id: Some(score_config.entity_id.clone()),
score_config_version: Some(score_config.version.to_string()),
score_config_name: Some(score_config.name.clone()),
data_type: score_config.data_type,
numeric_range: score_config.numeric_range.clone(),
categories: score_config.categories.clone(),
}
}
}
pub async fn run_llm_judge(
provider: &PreparedProvider,
scorer: &ScorerConfig,
score_config: &ScoreConfigInfo,
input_variables: &HashMap<String, Value>,
extra_metadata_fields: &[ExtraMetadataField],
include_reasoning: bool,
) -> Result<LlmJudgeOutput> {
let system_prompt = evaluator_system_prompt();
let user_prompt = render_template(&scorer.template, input_variables);
let prompt_messages = prompt_messages_json(&system_prompt, &user_prompt);
let output_schema = derive_output_schema(
score_config.data_type,
score_config.numeric_range.as_ref(),
score_config.categories.as_ref(),
extra_metadata_fields,
include_reasoning,
)?;
let call_params = ProviderRunParams {
model: scorer.params.model.clone(),
system_prompt,
user_prompt,
temperature: scorer.params.temperature.unwrap_or(0.0),
max_tokens: scorer.params.max_tokens.unwrap_or(256),
response_schema: output_schema.json_schema,
timeout_ms: scorer.params.timeout_ms.unwrap_or(30_000),
};
let start = Instant::now();
let result = provider.run(&call_params).await?;
let latency = start.elapsed();
let metadata = extract_metadata(&result.value, &output_schema.metadata_fields);
let reasoning = result
.value
.get(REASONING_FIELD)
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let value_field = &output_schema.value_field;
let (value_numeric, value_categorical, value_boolean) =
extract_score_value(&result.value, score_config.data_type, value_field);
Ok(LlmJudgeOutput {
value_numeric,
value_categorical,
value_boolean,
reasoning,
metadata,
raw_response: result.raw_response,
prompt_messages,
prompt_tokens: result.prompt_tokens,
completion_tokens: result.completion_tokens,
total_tokens: result.total_tokens,
model_used: result.model_used,
latency_ms: latency.as_millis() as i64,
})
}
pub(crate) fn build_prompt_messages_json(
scorer: &ScorerConfig,
input_variables: &HashMap<String, Value>,
) -> String {
let system_prompt = evaluator_system_prompt();
let user_prompt = render_template(&scorer.template, input_variables);
prompt_messages_json(&system_prompt, &user_prompt)
}
fn evaluator_system_prompt() -> String {
"You are an impartial LLM evaluator. Follow the instructions and output ONLY valid JSON."
.to_string()
}
fn prompt_messages_json(system_prompt: &str, user_prompt: &str) -> String {
serde_json::json!([
{ "role": "system", "content": system_prompt },
{ "role": "user", "content": user_prompt }
])
.to_string()
}
fn extract_metadata(value: &Value, metadata_fields: &[String]) -> Option<Value> {
if metadata_fields.is_empty() {
return None;
}
let mut meta = serde_json::json!({});
for field in metadata_fields {
if let Some(val) = value.get(field) {
meta[field] = val.clone();
}
}
if meta.as_object().map(|o| o.is_empty()).unwrap_or(true) {
None
} else {
Some(meta)
}
}
fn extract_score_value(
value: &Value,
data_type: ScoreConfigDataType,
value_field: &str,
) -> (Option<f64>, Option<String>, Option<bool>) {
match data_type {
ScoreConfigDataType::Numeric => {
(value.get(value_field).and_then(|v| v.as_f64()), None, None)
}
ScoreConfigDataType::Categorical => (
None,
value
.get(value_field)
.and_then(|v| v.as_str())
.map(|s| s.to_string()),
None,
),
ScoreConfigDataType::Boolean => {
(None, None, value.get(value_field).and_then(|v| v.as_bool()))
}
}
}
pub(super) fn render_template(template: &str, input_variables: &HashMap<String, Value>) -> String {
let mut result = template.to_string();
let mut start = 0;
while let Some(open) = result[start..].find("{{") {
let abs_open = start + open;
if let Some(close) = result[abs_open + 2..].find("}}") {
let abs_close = abs_open + 2 + close;
let variable = result[abs_open + 2..abs_close].trim();
let value = get_nested_attr(input_variables, variable);
result.replace_range(abs_open..abs_close + 2, &value);
start = abs_open + value.len();
} else {
break;
}
}
result
}
pub(super) fn get_nested_attr(attrs: &HashMap<String, Value>, path: &str) -> String {
let parts: Vec<&str> = path.split('.').collect();
let mut current = attrs.get(parts[0]);
for part in &parts[1..] {
current = current.and_then(|v| v.get(part));
}
match current {
Some(Value::String(s)) => s.clone(),
Some(v) => v.to_string(),
None => attrs
.get(path)
.map(|v| match v {
Value::String(s) => s.clone(),
v => v.to_string(),
})
.unwrap_or_default(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn extracts_numeric_score_from_fixed_score_field() {
let value = serde_json::json!({
"score": 0.75,
"reasoning": "grounded",
"failure_mode": "none"
});
let (numeric, categorical, boolean) =
extract_score_value(&value, ScoreConfigDataType::Numeric, "score");
let metadata = extract_metadata(&value, &["failure_mode".to_string()]);
assert_eq!(numeric, Some(0.75));
assert_eq!(categorical, None);
assert_eq!(boolean, None);
assert_eq!(metadata, Some(serde_json::json!({"failure_mode": "none"})));
}
#[test]
fn extracts_categorical_and_boolean_scores_from_fixed_score_field() {
let categorical = serde_json::json!({"score": "high"});
let boolean = serde_json::json!({"score": true});
assert_eq!(
extract_score_value(&categorical, ScoreConfigDataType::Categorical, "score"),
(None, Some("high".to_string()), None)
);
assert_eq!(
extract_score_value(&boolean, ScoreConfigDataType::Boolean, "score"),
(None, None, Some(true))
);
}
}
@@ -1,19 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
pub mod llm_judge;
pub mod registry;
pub mod remote_client;
pub mod schema_derivation;
@@ -1,304 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use chrono::Utc;
use config::ider;
use infra::table::{self, scorers::ScorerType};
use crate::common::{
meta::authz::Authz,
utils::auth::{is_ofga_object_visible, remove_ownership, set_ownership},
};
/// Errors that can occur when interacting with scorers.
#[derive(Debug, thiserror::Error)]
pub enum ScorerError {
#[error("InfraError# Internal error")]
InfraError(#[from] infra::errors::Error),
#[error("Scorer name cannot be empty")]
MissingName,
#[error("Scorer not found")]
NotFound,
#[error("Invalid scorer_type '{0}'. Must be one of: llm_judge, remote")]
InvalidScorerType(String),
#[error("scorer_type is immutable and cannot be changed on update")]
ScorerTypeImmutable,
#[error("produces_score is immutable and cannot be changed on update")]
ProducesScoreConfigIdImmutable,
#[error("Score config version not found")]
ScoreConfigVersionNotFound,
#[error("Invalid LLM Judge output schema: {0}")]
InvalidOutputSchema(String),
#[error("Scorer name already exists")]
DuplicateName,
#[error("Scorer cannot be deleted because it is used by one or more eval jobs")]
InUseByEvalJob,
}
#[tracing::instrument(skip(scorer))]
pub async fn save_scorer(
org_id: &str,
mut scorer: table::scorers::Scorer,
) -> Result<table::scorers::Scorer, ScorerError> {
scorer.name = scorer.name.trim().to_string();
if scorer.name.is_empty() {
return Err(ScorerError::MissingName);
}
if table::scorers::get_by_active_name(org_id, &scorer.name)
.await?
.is_some()
{
return Err(ScorerError::DuplicateName);
}
scorer.id = ider::generate();
scorer.entity_id = ider::generate();
scorer.org_id = org_id.to_string();
scorer.version = 1;
scorer.produces_score_config_version = resolve_score_config_version(org_id, &scorer).await?;
let now = Utc::now().timestamp_millis();
scorer.created_at = now;
scorer.updated_at = now;
scorer.is_active = true;
table::scorers::add(&scorer).await?;
set_ownership(org_id, "scorers", Authz::new(&scorer.entity_id)).await;
publish_scorer_put(&scorer).await;
Ok(scorer)
}
#[tracing::instrument(skip(scorer))]
pub async fn update_scorer(
org_id: &str,
entity_id: &str,
mut scorer: table::scorers::Scorer,
) -> Result<table::scorers::Scorer, ScorerError> {
let existing = table::scorers::get_by_entity_id(org_id, entity_id)
.await?
.ok_or(ScorerError::NotFound)?;
// scorer_type is immutable; produced score config can change with a new scorer version.
if scorer.scorer_type != existing.scorer_type {
return Err(ScorerError::ScorerTypeImmutable);
}
let produces_score_config_id = scorer
.produces_score_config_id
.clone()
.or_else(|| existing.produces_score_config_id.clone());
let new_name = scorer.name.trim().to_string();
let new_name = if new_name.is_empty() {
existing.name.clone()
} else {
new_name
};
if new_name != existing.name
&& table::scorers::get_by_active_name(org_id, &new_name)
.await?
.is_some()
{
return Err(ScorerError::DuplicateName);
}
let latest_version = table::scorers::get_latest_version(org_id, entity_id).await?;
scorer.id = ider::generate();
scorer.entity_id = existing.entity_id;
scorer.org_id = org_id.to_string();
scorer.name = new_name;
scorer.scorer_type = existing.scorer_type;
scorer.produces_score_config_id = produces_score_config_id;
scorer.produces_score_config_version = resolve_score_config_version(org_id, &scorer).await?;
scorer.version = latest_version + 1;
let now = Utc::now().timestamp_millis();
scorer.created_at = now;
scorer.updated_at = now;
scorer.is_active = true;
table::scorers::delete(entity_id, org_id).await?;
table::scorers::add(&scorer).await?;
publish_scorer_put(&scorer).await;
Ok(scorer)
}
async fn resolve_score_config_version(
org_id: &str,
scorer: &table::scorers::Scorer,
) -> Result<Option<i32>, ScorerError> {
match scorer.produces_score_config_id.as_deref() {
Some(entity_id) if !entity_id.is_empty() => {
if let Some(version) = scorer.produces_score_config_version {
let score_config =
table::score_configs::get_by_entity_id_and_version(org_id, entity_id, version)
.await?;
return match score_config {
Some(_) => Ok(Some(version)),
None => Err(ScorerError::ScoreConfigVersionNotFound),
};
}
table::score_configs::get_by_entity_id(org_id, entity_id)
.await?
.map(|score_config| Some(score_config.version))
.ok_or(ScorerError::ScoreConfigVersionNotFound)
}
_ => Ok(None),
}
}
#[tracing::instrument()]
pub async fn list_scorers(
org_id: &str,
scorer_type: Option<&ScorerType>,
permitted_objects: Option<Vec<String>>,
) -> Result<Vec<table::scorers::Scorer>, ScorerError> {
let scorers = match scorer_type {
Some(t) => table::scorers::get_by_type(org_id, t).await?,
None => table::scorers::get_all_by_org(org_id).await?,
};
Ok(scorers
.into_iter()
.filter(|scorer| {
is_ofga_object_visible(
org_id,
"scorer",
&scorer.entity_id,
permitted_objects.as_deref(),
)
})
.collect())
}
#[tracing::instrument()]
pub async fn get_scorer(
org_id: &str,
entity_id: &str,
) -> Result<table::scorers::Scorer, ScorerError> {
table::scorers::get_by_entity_id(org_id, entity_id)
.await?
.ok_or(ScorerError::NotFound)
}
#[tracing::instrument()]
pub async fn get_scorer_versions(
org_id: &str,
entity_id: &str,
) -> Result<Vec<table::scorers::Scorer>, ScorerError> {
let versions = table::scorers::get_versions(org_id, entity_id).await?;
Ok(versions)
}
#[tracing::instrument()]
pub async fn delete_scorer(org_id: &str, entity_id: &str) -> Result<(), ScorerError> {
table::scorers::get_by_entity_id(org_id, entity_id)
.await?
.ok_or(ScorerError::NotFound)?;
if table::online_eval_jobs::has_non_archived_by_scorer_ref(org_id, entity_id).await? {
return Err(ScorerError::InUseByEvalJob);
}
table::scorers::delete(entity_id, org_id).await?;
remove_ownership(org_id, "scorers", Authz::new(entity_id)).await;
publish_scorer_delete(org_id, entity_id).await;
Ok(())
}
#[cfg(feature = "enterprise")]
async fn publish_scorer_put(scorer: &table::scorers::Scorer) {
if !o2_enterprise::enterprise::common::config::get_config()
.super_cluster
.enabled
{
return;
}
let key = format!("/eval/scorers/{}/{}", scorer.org_id, scorer.entity_id);
let message = o2_enterprise::enterprise::super_cluster::queue::EvalScorerMessage::Put {
scorer: scorer.clone(),
};
match config::utils::json::to_vec(&message) {
Ok(value) => {
if let Err(e) =
o2_enterprise::enterprise::super_cluster::queue::eval_scorer_put(&key, value.into())
.await
{
log::error!("[Scorer] error publishing super cluster scorer put: {e}");
}
}
Err(e) => log::error!("[Scorer] error serializing scorer super cluster message: {e}"),
}
}
#[cfg(not(feature = "enterprise"))]
async fn publish_scorer_put(_scorer: &table::scorers::Scorer) {}
#[cfg(feature = "enterprise")]
async fn publish_scorer_delete(org_id: &str, entity_id: &str) {
if !o2_enterprise::enterprise::common::config::get_config()
.super_cluster
.enabled
{
return;
}
let key = format!("/eval/scorers/{}/{}", org_id, entity_id);
if let Err(e) = o2_enterprise::enterprise::super_cluster::queue::eval_scorer_delete(&key).await
{
log::error!("[Scorer] error publishing super cluster scorer delete: {e}");
}
}
#[cfg(not(feature = "enterprise"))]
async fn publish_scorer_delete(_org_id: &str, _entity_id: &str) {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_scorer_error_variants() {
let cases = vec![
ScorerError::MissingName,
ScorerError::NotFound,
ScorerError::InvalidScorerType("bad".to_string()),
ScorerError::ScorerTypeImmutable,
ScorerError::ProducesScoreConfigIdImmutable,
ScorerError::InvalidOutputSchema("bad".to_string()),
ScorerError::DuplicateName,
ScorerError::InUseByEvalJob,
];
for c in cases {
// each error should produce a non-empty display message
assert!(!c.to_string().is_empty());
}
}
#[test]
fn test_scorer_type_parse() {
assert_eq!("llm_judge".parse::<ScorerType>(), Ok(ScorerType::LlmJudge));
assert_eq!("remote".parse::<ScorerType>(), Ok(ScorerType::Remote));
assert!("code".parse::<ScorerType>().is_err());
assert!("".parse::<ScorerType>().is_err());
}
}
@@ -1,424 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Remote scorer HTTP client.
use std::{collections::HashMap, time::Duration};
use anyhow::Result;
use rand::RngExt;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteScorerOutput {
pub value: Value,
pub reasoning: Option<String>,
pub metadata: Option<Value>,
pub raw_response: String,
pub latency_ms: i64,
}
#[derive(Debug, Clone)]
pub struct RemoteScorerConfig {
pub endpoint: String,
pub http_method: String,
pub auth_type: String,
pub auth_token: Option<String>,
pub auth_username: Option<String>,
pub auth_password: Option<String>,
pub api_key_header: Option<String>,
pub custom_headers: Vec<(String, String)>,
pub content_type: String,
pub request_body_template: Option<String>,
pub timeout_ms: u64,
pub max_retries: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct RemoteScorerParams {
pub endpoint: String,
pub http_method: Option<String>,
pub auth: Option<RemoteScorerAuth>,
pub custom_headers: Option<Vec<RemoteScorerHeader>>,
pub content_type: Option<String>,
pub timeout_ms: Option<u64>,
pub max_retries: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
pub enum RemoteScorerAuth {
None,
Bearer { token: String },
Basic { username: String, password: String },
ApiKey { token: String, header_name: String },
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RemoteScorerHeader {
pub key: String,
pub value: String,
}
pub fn parse_remote_config(params: &Value) -> Result<RemoteScorerConfig> {
let params: RemoteScorerParams = serde_json::from_value(params.clone())
.map_err(|e| anyhow::anyhow!("Invalid remote scorer params: {e}"))?;
let (auth_type, auth_token, auth_username, auth_password, api_key_header) = match params.auth {
Some(RemoteScorerAuth::Bearer { token }) => {
("bearer".to_string(), Some(token), None, None, None)
}
Some(RemoteScorerAuth::Basic { username, password }) => (
"basic".to_string(),
None,
Some(username),
Some(password),
None,
),
Some(RemoteScorerAuth::ApiKey { token, header_name }) => (
"api_key".to_string(),
Some(token),
None,
None,
Some(header_name),
),
Some(RemoteScorerAuth::None) | None => ("none".to_string(), None, None, None, None),
};
Ok(RemoteScorerConfig {
endpoint: params.endpoint,
http_method: params
.http_method
.unwrap_or_else(|| "POST".to_string())
.to_uppercase(),
auth_type,
auth_token,
auth_username,
auth_password,
api_key_header,
custom_headers: params
.custom_headers
.unwrap_or_default()
.into_iter()
.map(|h| (h.key, h.value))
.collect(),
content_type: params
.content_type
.unwrap_or_else(|| "application/json".to_string()),
request_body_template: None,
timeout_ms: params.timeout_ms.unwrap_or(30_000),
max_retries: params.max_retries.unwrap_or(3),
})
}
pub async fn run_remote_scorer(
config: &RemoteScorerConfig,
span_attributes: &HashMap<String, Value>,
) -> Result<RemoteScorerOutput> {
let start = std::time::Instant::now();
let body_template = config.request_body_template.as_deref().unwrap_or("{}");
let rendered_body = render_template(body_template, span_attributes);
let mut last_error: Option<anyhow::Error> = None;
for attempt in 0..=config.max_retries {
if attempt > 0 {
tokio::time::sleep(Duration::from_millis(compute_backoff_ms(attempt))).await;
}
match execute_request(config, &rendered_body).await {
Ok((output, raw_response)) => {
let latency = start.elapsed();
let parsed = parse_remote_response(output)?;
return Ok(RemoteScorerOutput {
value: parsed.value,
reasoning: parsed.reason,
metadata: parsed.metadata,
raw_response,
latency_ms: latency.as_millis() as i64,
});
}
Err(e) => {
log::warn!(
"[REMOTE-SCORER] Attempt {}/{} failed for {}: {}",
attempt + 1,
config.max_retries + 1,
config.endpoint,
e
);
last_error = Some(e);
}
}
}
Err(last_error.unwrap_or_else(|| anyhow::anyhow!("remote scorer request failed")))
}
fn render_template(template: &str, attrs: &HashMap<String, Value>) -> String {
let mut result = template.to_string();
let mut start = 0;
while let Some(open) = result[start..].find("{{") {
let abs_open = start + open;
if let Some(close) = result[abs_open..].find("}}") {
let abs_close = abs_open + close + 2;
let path = result[abs_open + 2..abs_close - 2].trim();
let mut value = get_attr(attrs, path);
if is_inside_string_literal(&result, abs_open) {
value = escape_json_string_content(&value);
}
result.replace_range(abs_open..abs_close, &value);
start = abs_open + value.len();
} else {
break;
}
}
result
}
fn get_attr(attrs: &HashMap<String, Value>, path: &str) -> String {
let parts: Vec<&str> = path.split('.').collect();
let mut current = attrs.get(parts[0]);
for part in &parts[1..] {
current = current.and_then(|v| v.get(part));
}
match current {
Some(Value::String(s)) => s.clone(),
Some(v) => v.to_string(),
None => path.to_string(),
}
}
fn is_inside_string_literal(template: &str, pos: usize) -> bool {
let mut in_string = false;
let mut escaped = false;
for ch in template[..pos].chars() {
if escaped {
escaped = false;
continue;
}
match ch {
'\\' => escaped = true,
'"' => in_string = !in_string,
_ => {}
}
}
in_string
}
fn escape_json_string_content(value: &str) -> String {
serde_json::to_string(value)
.unwrap_or_else(|_| value.to_string())
.trim_matches('"')
.to_string()
}
async fn execute_request(config: &RemoteScorerConfig, body: &str) -> Result<(Value, String)> {
let client = reqwest::Client::builder()
.timeout(Duration::from_millis(config.timeout_ms))
.build()?;
let mut req = match config.http_method.as_str() {
"GET" => client.get(&config.endpoint),
"PUT" => client.put(&config.endpoint),
"PATCH" => client.patch(&config.endpoint),
"DELETE" => client.delete(&config.endpoint),
_ => client.post(&config.endpoint),
};
match config.auth_type.as_str() {
"bearer" => {
if let Some(token) = &config.auth_token {
req = req.header("Authorization", format!("Bearer {token}"));
}
}
"basic" => {
if let (Some(user), Some(pass)) = (&config.auth_username, &config.auth_password) {
let encoded = base64::Engine::encode(
&base64::engine::general_purpose::STANDARD,
format!("{user}:{pass}"),
);
req = req.header("Authorization", format!("Basic {encoded}"));
}
}
"api_key" => {
if let (Some(header_name), Some(token)) = (&config.api_key_header, &config.auth_token) {
req = req.header(header_name, token);
}
}
_ => {}
}
for (key, val) in &config.custom_headers {
req = req.header(key, val);
}
if !body.is_empty() && config.http_method != "GET" {
req = req.header("Content-Type", &config.content_type);
req = req.body(body.to_string());
}
let resp = req.send().await?;
let status = resp.status();
let raw_body = resp.text().await?;
let parsed = serde_json::from_str(&raw_body).map_err(|e| {
anyhow::anyhow!(
"Failed to parse remote scorer response as JSON (HTTP {}): {e}",
status
)
})?;
Ok((parsed, raw_body))
}
#[derive(Debug)]
struct ParsedRemoteResponse {
value: Value,
reason: Option<String>,
metadata: Option<Value>,
}
fn parse_remote_response(response: Value) -> Result<ParsedRemoteResponse> {
let mut obj = match response {
Value::Object(obj) => obj,
_ => anyhow::bail!("Remote scorer response must be a JSON object"),
};
let code = obj
.remove("code")
.and_then(|v| v.as_str().map(|s| s.to_string()))
.ok_or_else(|| anyhow::anyhow!("Remote scorer response missing string field 'code'"))?;
let reason = obj
.remove("reason")
.and_then(|v| v.as_str().map(|s| s.to_string()));
if code != "OK" {
anyhow::bail!(
"Remote scorer returned code '{}': {}",
code,
reason.as_deref().unwrap_or("no reason provided")
);
}
let value = obj.remove("value").ok_or_else(|| {
anyhow::anyhow!("Remote scorer response missing field 'value' for code OK")
})?;
Ok(ParsedRemoteResponse {
value,
reason,
metadata: object_to_metadata(obj),
})
}
fn object_to_metadata(obj: Map<String, Value>) -> Option<Value> {
if obj.is_empty() {
None
} else {
Some(Value::Object(obj))
}
}
fn compute_backoff_ms(attempt: u32) -> u64 {
let base = 200u64.saturating_mul(1u64 << (attempt - 1).min(4));
let mut rng = rand::rng();
base + rng.random_range(0..=base / 2)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_parse_remote_response_ok_with_metadata() {
let parsed = parse_remote_response(serde_json::json!({
"code": "OK",
"value": 0.85,
"reason": "good",
"confidence": "high"
}))
.unwrap();
assert_eq!(parsed.value, serde_json::json!(0.85));
assert_eq!(parsed.reason.as_deref(), Some("good"));
assert_eq!(parsed.metadata.unwrap()["confidence"], "high");
}
#[test]
fn test_parse_remote_response_error_code_fails() {
let err = parse_remote_response(serde_json::json!({
"code": "INPUT_TOO_LONG",
"reason": "too long"
}))
.unwrap_err();
assert!(err.to_string().contains("INPUT_TOO_LONG"));
}
#[test]
fn test_parse_remote_response_ok_requires_value() {
let err = parse_remote_response(serde_json::json!({
"code": "OK",
"reason": "missing value"
}))
.unwrap_err();
assert!(err.to_string().contains("missing field 'value'"));
}
#[test]
fn test_render_template_escapes_values_inside_json_strings() {
let attrs = HashMap::from([
(
"input".to_string(),
serde_json::json!(r#"[{"content":"hello \"world\"","role":"user"}]"#),
),
("output".to_string(), serde_json::json!("line 1\nline 2")),
]);
let result = render_template(r#"{"input":"{{ input }}","output":"{{ output }}"}"#, &attrs);
let parsed: Value = serde_json::from_str(&result).unwrap();
assert_eq!(parsed["input"], attrs["input"]);
assert_eq!(parsed["output"], attrs["output"]);
}
#[test]
fn test_render_template_preserves_free_text_values() {
let attrs = HashMap::from([(
"span".to_string(),
serde_json::json!({"name": "test-span", "kind": "internal"}),
)]);
let result = render_template("Span: {{ span.name }} ({{ span.kind }})", &attrs);
assert_eq!(result, "Span: test-span (internal)");
}
#[test]
fn test_remote_params_reject_response_paths() {
let err = parse_remote_config(&serde_json::json!({
"endpoint": "http://localhost:8000/evaluate",
"response_value_path": "value"
}))
.unwrap_err();
assert!(err.to_string().contains("response_value_path"));
}
}
@@ -1,284 +0,0 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Schema derivation for LLM Judge output.
use anyhow::Result;
use infra::table::score_configs::ScoreConfigDataType;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use utoipa::ToSchema;
#[derive(Debug, Clone)]
pub struct DerivedSchema {
pub json_schema: Value,
pub value_field: String,
pub metadata_fields: Vec<String>,
}
pub const SCORE_FIELD: &str = "score";
pub const REASONING_FIELD: &str = "reasoning";
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum ExtraMetadataFieldType {
#[default]
String,
Number,
Boolean,
}
impl ExtraMetadataFieldType {
fn json_schema_type(self) -> &'static str {
match self {
Self::String => "string",
Self::Number => "number",
Self::Boolean => "boolean",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(deny_unknown_fields)]
pub struct ExtraMetadataField {
pub name: String,
#[serde(rename = "type")]
pub data_type: ExtraMetadataFieldType,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
}
pub fn derive_output_schema(
data_type: ScoreConfigDataType,
numeric_range: Option<&Value>,
categories: Option<&Value>,
extra_metadata_fields: &[ExtraMetadataField],
include_reasoning: bool,
) -> Result<DerivedSchema> {
let value_schema = match data_type {
ScoreConfigDataType::Numeric => {
let mut schema = serde_json::json!({
"type": "number",
"description": "The numeric score value"
});
if let Some(range) = numeric_range
&& let (Some(min), Some(max)) = (range["min"].as_f64(), range["max"].as_f64())
{
schema["minimum"] = serde_json::json!(min);
schema["maximum"] = serde_json::json!(max);
}
schema
}
ScoreConfigDataType::Categorical => {
let allowed: Vec<&str> = categories
.and_then(|c| c.as_array())
.map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
.unwrap_or_default();
if allowed.is_empty() {
anyhow::bail!("categories must be provided for categorical data_type");
}
serde_json::json!({
"type": "string",
"enum": allowed,
"description": "The categorical score value"
})
}
ScoreConfigDataType::Boolean => serde_json::json!({
"type": "boolean",
"description": "The boolean pass/fail result"
}),
};
let mut properties = serde_json::json!({});
properties[SCORE_FIELD] = value_schema;
if include_reasoning {
properties[REASONING_FIELD] = serde_json::json!({
"type": "string",
"description": "Brief explanation of the score"
});
}
let mut metadata_fields = Vec::new();
for field in extra_metadata_fields {
let sanitized = field
.name
.chars()
.map(|c| {
if c.is_alphanumeric() || c == '_' {
c
} else {
'_'
}
})
.collect::<String>();
if !sanitized.is_empty() && sanitized != SCORE_FIELD && sanitized != REASONING_FIELD {
metadata_fields.push(sanitized.clone());
let description = field
.description
.as_deref()
.filter(|description| !description.trim().is_empty())
.map(str::to_string)
.unwrap_or_else(|| format!("Additional metadata: {}", field.name));
properties[&sanitized] = serde_json::json!({
"type": field.data_type.json_schema_type(),
"description": description
});
}
}
let mut required_fields = vec![SCORE_FIELD.to_string()];
if include_reasoning {
required_fields.push(REASONING_FIELD.to_string());
}
Ok(DerivedSchema {
json_schema: serde_json::json!({
"type": "object",
"properties": properties,
"required": required_fields,
"additionalProperties": false
}),
value_field: SCORE_FIELD.to_string(),
metadata_fields,
})
}
#[cfg(test)]
mod tests {
use super::*;
fn metadata_field(
name: &str,
data_type: ExtraMetadataFieldType,
description: Option<&str>,
) -> ExtraMetadataField {
ExtraMetadataField {
name: name.to_string(),
data_type,
description: description.map(str::to_string),
}
}
#[test]
fn derives_numeric_score_schema_from_range() {
let schema = derive_output_schema(
ScoreConfigDataType::Numeric,
Some(&serde_json::json!({"min": 0.0, "max": 1.0})),
None,
&[metadata_field(
"failure_mode",
ExtraMetadataFieldType::String,
Some("Reason for low score"),
)],
true,
)
.unwrap();
assert_eq!(schema.value_field, "score");
assert_eq!(schema.json_schema["properties"]["score"]["type"], "number");
assert_eq!(schema.json_schema["properties"]["score"]["minimum"], 0.0);
assert_eq!(schema.json_schema["properties"]["score"]["maximum"], 1.0);
assert_eq!(
schema.json_schema["required"],
serde_json::json!(["score", "reasoning"])
);
assert_eq!(schema.metadata_fields, vec!["failure_mode".to_string()]);
assert_eq!(
schema.json_schema["properties"]["failure_mode"],
serde_json::json!({
"type": "string",
"description": "Reason for low score"
})
);
}
#[test]
fn derives_categorical_score_schema_from_categories() {
let schema = derive_output_schema(
ScoreConfigDataType::Categorical,
None,
Some(&serde_json::json!(["low", "medium", "high"])),
&[],
false,
)
.unwrap();
assert_eq!(schema.json_schema["properties"]["score"]["type"], "string");
assert_eq!(
schema.json_schema["properties"]["score"]["enum"],
serde_json::json!(["low", "medium", "high"])
);
assert_eq!(schema.json_schema["required"], serde_json::json!(["score"]));
}
#[test]
fn derives_boolean_score_schema() {
let schema = derive_output_schema(
ScoreConfigDataType::Boolean,
None,
None,
&[
metadata_field("score", ExtraMetadataFieldType::String, None),
metadata_field("reasoning", ExtraMetadataFieldType::String, None),
metadata_field("debug flag", ExtraMetadataFieldType::Boolean, None),
],
true,
)
.unwrap();
assert_eq!(schema.json_schema["properties"]["score"]["type"], "boolean");
assert_eq!(schema.metadata_fields, vec!["debug_flag".to_string()]);
assert_eq!(
schema.json_schema["properties"]["debug_flag"],
serde_json::json!({
"type": "boolean",
"description": "Additional metadata: debug flag"
})
);
}
#[test]
fn rejects_string_metadata_field_entries() {
let result: Result<ExtraMetadataField, _> =
serde_json::from_value(serde_json::json!("confidence"));
assert!(result.is_err());
}
#[test]
fn derives_number_metadata_field_schema() {
let schema = derive_output_schema(
ScoreConfigDataType::Numeric,
None,
None,
&[metadata_field(
"confidence",
ExtraMetadataFieldType::Number,
Some("Judge confidence from 0.0 to 1.0"),
)],
true,
)
.unwrap();
assert_eq!(
schema.json_schema["properties"]["confidence"],
serde_json::json!({
"type": "number",
"description": "Judge confidence from 0.0 to 1.0"
})
);
}
}
+2
View File
@@ -36,6 +36,7 @@ pub mod grpc;
pub mod ingestion;
pub mod ingestion_tokens;
pub mod kv;
#[cfg(feature = "enterprise")]
pub mod llm_evaluations;
pub mod logs;
pub mod metadata;
@@ -48,6 +49,7 @@ pub mod org_usage;
pub mod organization;
pub mod pipeline;
pub mod promql;
#[cfg(feature = "enterprise")]
pub mod providers;
#[cfg(feature = "enterprise")]
pub mod ratelimit;
+31 -1
View File
@@ -13,9 +13,10 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#[cfg(feature = "enterprise")]
use std::hash::{Hash, Hasher};
use std::{
collections::{HashMap, HashSet},
hash::{Hash, Hasher},
sync::LazyLock as Lazy,
};
@@ -65,6 +66,7 @@ struct BatchBuffer {
last_write: Instant,
}
#[cfg(feature = "enterprise")]
fn should_sample_eval_record(record: &json::Value, idx: usize, sampling_rate: f64) -> bool {
if sampling_rate >= 1.0 {
return true;
@@ -1341,7 +1343,22 @@ async fn process_node(
);
}
}
#[cfg(feature = "enterprise")]
NodeData::LlmEvaluation(params) => {
if !o2_enterprise::enterprise::common::config::get_config()
.common
.online_evals_enabled
{
log::warn!(
"[Pipeline]: LLM evaluation node {node_idx} skipped because online evals are disabled"
);
while receiver.recv().await.is_some() {
count += 1;
}
log::info!("[Pipeline]: LLM evaluation node {node_idx} skipped {count} records");
return Ok(());
}
log::info!("[Pipeline]: LLM evaluation node {node_idx} starts processing");
if let Err(e) =
@@ -1470,6 +1487,19 @@ async fn process_node(
"[Pipeline]: LLM evaluation node {node_idx} done processing {count} records"
);
}
#[cfg(not(feature = "enterprise"))]
NodeData::LlmEvaluation(_) => {
log::warn!(
"[Pipeline]: LLM evaluation node {node_idx} skipped because online evals are enterprise-only"
);
let mut skipped_count = 0usize;
while receiver.recv().await.is_some() {
skipped_count += 1;
}
log::info!(
"[Pipeline]: LLM evaluation node {node_idx} skipped {skipped_count} records"
);
}
}
// all cloned senders dropped when function goes out of scope -> close the channel
+3
View File
@@ -83,6 +83,9 @@ pub async fn update_provider(
let existing = table::providers::get(provider_id)
.await?
.ok_or(ProviderError::NotFound)?;
if existing.org_id != org_id {
return Err(ProviderError::NotFound);
}
// Check name uniqueness excluding self
let all = table::providers::get_all_by_org(org_id).await?;
+4
View File
@@ -42,17 +42,21 @@ use tokio::sync::oneshot;
#[cfg(feature = "cloud")]
pub mod cloud_events;
#[cfg(feature = "enterprise")]
mod evaluator_schema;
mod ingestion;
#[cfg(feature = "enterprise")]
mod llm_scores_schema;
mod queues;
pub mod search;
mod triggers_schema;
mod usage_schema;
#[cfg(feature = "enterprise")]
pub(crate) use evaluator_schema::ensure_evaluator_stream_initialized;
#[cfg(feature = "cloud")]
pub use ingestion::ingest_data_retention_usages;
#[cfg(feature = "enterprise")]
pub(crate) use llm_scores_schema::ensure_llm_scores_stream_initialized;
pub async fn run() {
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -329,7 +329,7 @@ export default defineComponent({
});
// Backend `/config` flag `online_evals_enabled` — controlled by
// `ZO_ONLINE_EVALS_ENABLED`. Reactive so the menu picks it up regardless
// enterprise `O2_ONLINE_EVALS_ENABLED`. Reactive so the menu picks it up regardless
// of whether the config response arrived before or after this component
// mounted.
const isOnlineEvalsEnabled = computed(() => {