Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 48 additions & 15 deletions scripts/ingest.ps1
Original file line number Diff line number Diff line change
Expand Up @@ -19,17 +19,23 @@ $ProgressPreference = 'SilentlyContinue'
$COLLECTOR_VERSION = "0.157.0"
$INSTALL_DIR = "$env:LOCALAPPDATA\parseable-otelcol"
$COLLECTOR_EXE = "$INSTALL_DIR\otelcol.exe"
$CONFIG_FILE = "$PSScriptRoot\otelcol.yaml"
$PID_FILE = "$PSScriptRoot\otelcol.pid"
$LOG_FILE = "$PSScriptRoot\otelcol.log"
$ERROR_LOG_FILE = "$PSScriptRoot\otelcol.err.log"
$SCRIPT_DIR = if ([string]::IsNullOrWhiteSpace($PSScriptRoot)) {
(Get-Location).Path
}
else {
$PSScriptRoot
}
$CONFIG_FILE = Join-Path $SCRIPT_DIR "otelcol.yaml"
$PID_FILE = Join-Path $SCRIPT_DIR "otelcol.pid"
$LOG_FILE = Join-Path $SCRIPT_DIR "otelcol.log"
$ERROR_LOG_FILE = Join-Path $SCRIPT_DIR "otelcol.err.log"

$SCRIPT_PATH = $PSCommandPath
if ([string]::IsNullOrWhiteSpace($SCRIPT_PATH)) {
$SCRIPT_PATH = $MyInvocation.MyCommand.Path
}
if ([string]::IsNullOrWhiteSpace($SCRIPT_PATH)) {
$SCRIPT_PATH = Join-Path $PSScriptRoot "ingest.ps1"
$SCRIPT_PATH = Join-Path $SCRIPT_DIR "ingest.ps1"
Comment thread
nikhilsinhaparseable marked this conversation as resolved.
}
$SCRIPT_CMD = "powershell -NoProfile -ExecutionPolicy Bypass -File '$SCRIPT_PATH'"

Expand All @@ -48,6 +54,11 @@ function Write-ErrorMsg {
Write-Host "[ERROR] $Message" -ForegroundColor Red
}

function ConvertTo-PowerShellSingleQuoted {
param([string]$Value)
return "'" + $Value.Replace("'", "''") + "'"
}

function Write-ParseableBanner {
$accent = "$([char]27)[38;2;158;158;240m"
$reset = "$([char]27)[0m"
Expand Down Expand Up @@ -135,9 +146,19 @@ function Stop-Collector {

try {
Stop-Process -Id $processId -Force -ErrorAction Stop
Start-Sleep -Seconds 2
for ($attempt = 0; $attempt -lt 10; $attempt++) {
if ($null -eq (Get-Process -Id $processId -ErrorAction SilentlyContinue)) {
break
}
Start-Sleep -Seconds 1
}

if ($null -ne (Get-Process -Id $processId -ErrorAction SilentlyContinue)) {
throw "Process $processId did not stop within 10 seconds; PID file was preserved."
}

Remove-Item $PID_FILE -Force -ErrorAction Stop
Write-Info "OpenTelemetry Collector stopped successfully"
Remove-Item $PID_FILE -ErrorAction SilentlyContinue
}
catch {
Write-ErrorMsg "Failed to stop OpenTelemetry Collector: $_"
Expand All @@ -158,8 +179,10 @@ function Show-Status {
Write-Info "Config file: $CONFIG_FILE"
Write-Info "Log file: $LOG_FILE"
Write-Info "Error log file: $ERROR_LOG_FILE"
Write-Info "To see logs: $SCRIPT_CMD logs"
Write-Info "To stop: $SCRIPT_CMD stop"
$pidFileCommand = ConvertTo-PowerShellSingleQuoted $PID_FILE
$errorLogFileCommand = ConvertTo-PowerShellSingleQuoted $ERROR_LOG_FILE
Write-Info "To see logs: Get-Content $errorLogFileCommand -Tail 80 -Wait"
Write-Info "To stop: Stop-Process -Id (Get-Content $pidFileCommand); Remove-Item $pidFileCommand"
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
else {
Write-Warning "OpenTelemetry Collector is not running"
Expand Down Expand Up @@ -292,15 +315,18 @@ function Start-Collector {
$stillRunning = Get-Process -Id $process.Id -ErrorAction SilentlyContinue
if (-not $stillRunning) {
Write-ErrorMsg "OpenTelemetry Collector exited immediately"
Write-ErrorMsg "Run '$SCRIPT_CMD logs' to see error details"
$errorLogFileCommand = ConvertTo-PowerShellSingleQuoted $ERROR_LOG_FILE
Write-ErrorMsg "Check logs: Get-Content $errorLogFileCommand -Tail 80"
Remove-Item $PID_FILE -ErrorAction SilentlyContinue
exit 1
}

$pidFileCommand = ConvertTo-PowerShellSingleQuoted $PID_FILE
$errorLogFileCommand = ConvertTo-PowerShellSingleQuoted $ERROR_LOG_FILE
Write-Info "OpenTelemetry Collector started successfully (PID: $($process.Id))"
Write-Info "To check status: $SCRIPT_CMD status"
Write-Info "To see logs: $SCRIPT_CMD logs"
Write-Info "To stop: $SCRIPT_CMD stop"
Write-Info "To check status: Get-Process -Id (Get-Content $pidFileCommand)"
Write-Info "To see logs: Get-Content $errorLogFileCommand -Tail 80 -Wait"
Write-Info "To stop: Stop-Process -Id (Get-Content $pidFileCommand); Remove-Item $pidFileCommand"
return $true
}

Expand Down Expand Up @@ -432,7 +458,9 @@ function Setup-Collector {

$configContent = ($configLines -join [Environment]::NewLine) + [Environment]::NewLine
$utf8NoBom = New-Object System.Text.UTF8Encoding $false
$tempConfigFile = "$CONFIG_FILE.$([guid]::NewGuid().ToString('N')).tmp"
$configUpdateId = [guid]::NewGuid().ToString('N')
$tempConfigFile = "$CONFIG_FILE.$configUpdateId.tmp"
$backupConfigFile = "$CONFIG_FILE.$configUpdateId.bak"

try {
$tempConfigStream = [System.IO.File]::Create($tempConfigFile)
Expand Down Expand Up @@ -462,15 +490,20 @@ function Setup-Collector {

if (Test-Path $CONFIG_FILE) {
Set-Acl -Path $CONFIG_FILE -AclObject $acl -ErrorAction Stop
[System.IO.File]::Replace($tempConfigFile, $CONFIG_FILE, $null)
[System.IO.File]::Replace($tempConfigFile, $CONFIG_FILE, $backupConfigFile)
}
else {
[System.IO.File]::Move($tempConfigFile, $CONFIG_FILE)
}
Set-Acl -Path $CONFIG_FILE -AclObject $acl -ErrorAction Stop
}
catch {
Write-ErrorMsg "Failed to update OpenTelemetry Collector configuration: $_"
exit 1
}
finally {
Remove-Item $tempConfigFile -Force -ErrorAction SilentlyContinue
Remove-Item $backupConfigFile -Force -ErrorAction SilentlyContinue
}

if (Test-CollectorRunning) {
Expand Down
32 changes: 26 additions & 6 deletions scripts/ingest.sh
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,10 @@ stop_collector() {
if is_running; then
PID=$(cat "$PID_FILE")
print_info "Stopping OpenTelemetry Collector (PID: $PID)..."
kill "$PID"
if ! kill "$PID"; then
print_error "Failed to stop OpenTelemetry Collector; PID file was preserved"
return 1
fi

for _ in {1..10}; do
if ! ps -p "$PID" > /dev/null 2>&1; then
Expand All @@ -93,11 +96,28 @@ stop_collector() {
sleep 1
done

if ps -p "$PID" > /dev/null 2>&1; then
if is_running; then
print_warning "Force killing OpenTelemetry Collector..."
kill -9 "$PID"
rm -f "$PID_FILE"
if ! kill -9 "$PID"; then
print_error "Failed to force stop OpenTelemetry Collector; PID file was preserved"
return 1
fi

for _ in {1..5}; do
if ! ps -p "$PID" > /dev/null 2>&1; then
rm -f "$PID_FILE"
print_info "✓ OpenTelemetry Collector stopped successfully"
return 0
fi
sleep 1
done

print_error "OpenTelemetry Collector did not stop; PID file was preserved"
return 1
fi

rm -f "$PID_FILE"
print_info "✓ OpenTelemetry Collector stopped successfully"
else
print_warning "OpenTelemetry Collector is not running"
if [ -f "$PID_FILE" ]; then
Expand Down Expand Up @@ -256,8 +276,8 @@ start_collector() {
if ps -p "$PID" > /dev/null 2>&1; then
print_info "✓ OpenTelemetry Collector started successfully (PID: $PID)"
print_info "View logs: tail -f $LOG_FILE"
print_info "Check status: $0 status"
print_info "Stop: $0 stop"
print_info "Check status: ps -p \$(cat $PID_FILE) -o pid,ppid,user,%cpu,%mem,etime,command"
Comment thread
coderabbitai[bot] marked this conversation as resolved.
print_info "Stop: kill \$(cat $PID_FILE) && rm -f $PID_FILE"
else
print_error "✗ OpenTelemetry Collector failed to start. Check logs: cat $LOG_FILE"
rm -f "$PID_FILE"
Expand Down
30 changes: 23 additions & 7 deletions src/alerts/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,10 @@ pub fn create_default_alerts_manager() -> Alerts {
alerts
}

fn alert_tenant_from_storage_key(tenant: &str) -> Option<String> {
(!tenant.is_empty() && tenant != DEFAULT_TENANT).then(|| tenant.to_owned())
}

pub async fn user_auth_for_alert_config(
session: &SessionKey,
alert: &AlertConfig,
Expand Down Expand Up @@ -1174,18 +1178,14 @@ impl AlertManagerTrait for Alerts {
let mut map = self.alerts.write().await;

for (tenant_id, raw_bytes) in raw_objects {
let tenant = if tenant_id.is_empty() {
&None
} else {
&Some(tenant_id.clone())
};
let tenant = alert_tenant_from_storage_key(&tenant_id);
for alert_bytes in raw_bytes {
let Some(mut alert) = parse_alert_config(&alert_bytes, tenant).await else {
let Some(mut alert) = parse_alert_config(&alert_bytes, &tenant).await else {
continue;
};

// ensure that alert config's tenant is correctly set
alert.tenant_id.clone_from(tenant);
alert.tenant_id.clone_from(&tenant);

let alert = match alert_from_config_oss(alert) {
Ok(alert) => alert,
Expand Down Expand Up @@ -1673,3 +1673,19 @@ fn get_severity_priority(severity: &Severity) -> u8 {
Severity::Low => 3,
}
}

#[cfg(test)]
mod tests {
use super::alert_tenant_from_storage_key;
use crate::parseable::DEFAULT_TENANT;

#[test]
fn default_storage_tenant_is_not_an_explicit_tenant() {
assert_eq!(alert_tenant_from_storage_key(""), None);
assert_eq!(alert_tenant_from_storage_key(DEFAULT_TENANT), None);
assert_eq!(
alert_tenant_from_storage_key("tenant-a"),
Some("tenant-a".to_owned())
);
}
}
116 changes: 106 additions & 10 deletions src/metastore/metastores/object_store_metastore.rs
Original file line number Diff line number Diff line change
Expand Up @@ -325,11 +325,17 @@ impl Metastore for ObjectStoreMetastore {
obj: &dyn MetastoreObject,
tenant_id: &Option<String>,
) -> Result<(), MetastoreError> {
let path = obj.get_object_path();
Ok(self
.storage
.delete_object(&RelativePathBuf::from(path), tenant_id)
.await?)
let id = Ulid::from_string(&obj.get_object_id()).map_err(|e| MetastoreError::Error {
status_code: StatusCode::BAD_REQUEST,
message: e.to_string(),
flow: "delete_alert".into(),
})?;
let path = alert_json_path(id, tenant_id);
match self.storage.delete_object(&path, tenant_id).await {
Ok(()) => Ok(()),
Err(err) if is_missing_optional_dir(&err) => Ok(()),
Err(err) => Err(MetastoreError::ObjectStorageError(err)),
}
}

/// alerts state
Expand Down Expand Up @@ -439,11 +445,17 @@ impl Metastore for ObjectStoreMetastore {
obj: &dyn MetastoreObject,
tenant_id: &Option<String>,
) -> Result<(), MetastoreError> {
let path = obj.get_object_path();
Ok(self
.storage
.delete_object(&RelativePathBuf::from(path), tenant_id)
.await?)
let id = Ulid::from_string(&obj.get_object_id()).map_err(|e| MetastoreError::Error {
status_code: StatusCode::BAD_REQUEST,
message: e.to_string(),
flow: "delete_alert_state".into(),
})?;
let path = alert_state_json_path(id, tenant_id);
match self.storage.delete_object(&path, tenant_id).await {
Ok(()) => Ok(()),
Err(err) if is_missing_optional_dir(&err) => Ok(()),
Err(err) => Err(MetastoreError::ObjectStorageError(err)),
}
}

/// Get MTTR history from storage
Expand Down Expand Up @@ -1506,3 +1518,87 @@ impl Metastore for ObjectStoreMetastore {
}
}
}

#[cfg(test)]
mod tests {
use super::*;
use crate::storage::LocalFS;

#[derive(serde::Serialize)]
struct TestMetastoreObject {
id: Ulid,
path: String,
}

impl MetastoreObject for TestMetastoreObject {
fn get_object_path(&self) -> String {
self.path.clone()
}

fn get_object_id(&self) -> String {
self.id.to_string()
}
}

#[tokio::test]
async fn delete_alert_uses_request_tenant_path_and_is_idempotent() {
let temp_dir = tempfile::tempdir().unwrap();
let storage = Arc::new(LocalFS::new(temp_dir.path().to_path_buf()));
let metastore = ObjectStoreMetastore {
storage: storage.clone(),
};
let id = Ulid::new();
let tenant_id = None;
let path = alert_json_path(id, &tenant_id);
storage
.put_object(&path, Bytes::from_static(b"{}"), &tenant_id)
.await
.unwrap();
let object = TestMetastoreObject {
id,
path: format!("{DEFAULT_TENANT}/{ALERTS_ROOT_DIRECTORY}/{id}.json"),
};

metastore.delete_alert(&object, &tenant_id).await.unwrap();
metastore.delete_alert(&object, &tenant_id).await.unwrap();

assert!(matches!(
storage.get_object(&path, &tenant_id).await,
Err(ObjectStorageError::NoSuchKey(_))
));
}

#[tokio::test]
async fn delete_alert_state_uses_request_tenant_path_and_is_idempotent() {
let temp_dir = tempfile::tempdir().unwrap();
let storage = Arc::new(LocalFS::new(temp_dir.path().to_path_buf()));
let metastore = ObjectStoreMetastore {
storage: storage.clone(),
};
let id = Ulid::new();
let tenant_id = None;
let path = alert_state_json_path(id, &tenant_id);
storage
.put_object(&path, Bytes::from_static(b"{}"), &tenant_id)
.await
.unwrap();
let object = TestMetastoreObject {
id,
path: format!("{DEFAULT_TENANT}/{ALERTS_ROOT_DIRECTORY}/alert_state_{id}.json"),
};

metastore
.delete_alert_state(&object, &tenant_id)
.await
.unwrap();
metastore
.delete_alert_state(&object, &tenant_id)
.await
.unwrap();

assert!(matches!(
storage.get_object(&path, &tenant_id).await,
Err(ObjectStorageError::NoSuchKey(_))
));
}
}
2 changes: 2 additions & 0 deletions src/storage/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,8 @@ mod azure_blob;
pub mod field_stats;
mod gcs;
mod localfs;
#[cfg(test)]
pub(crate) use localfs::LocalFS;
mod metrics_layer;
pub mod object_storage;
pub mod retention;
Expand Down
Loading