mirror of
https://github.com/OPM/ResInsight.git
synced 2026-10-27 07:30:20 -05:00
Magne Sjaastad
e2319d0bd3
Send one batched request per event batch instead of one request per event. Application Insights v2/track accepts an array of envelopes, so a batch of up to maxBatchSize (512) events was previously issued as 512 concurrent POSTs. QNetworkAccessManager only opens a handful of connections per host, so the queued requests were aborted by their own transfer timeout before reaching the wire, reported as "HTTP 0: Operation canceled". Honour the circuit breaker in processEvents(). It was only checked when enqueuing, so the up to 10000 already queued events kept being posted at full rate after the breaker opened. The crash and shutdown flush bypasses the breaker so a crash report is still attempted. Keep at most one request in flight. Failures are only observed when the reply finishes, which for an unreachable endpoint takes the full transfer timeout, and the 100 ms timer kept launching requests during that window. Log the circuit breaker transition once instead of on every subsequent failure. Only read the reply body when the device is open. An aborted reply has a closed device, and reading from it emitted a QIODevice warning per request. Call attemptReconnection() from the process timer. It was dead code, so once the breaker opened it could only close if a reply happened to succeed.
978 lines
37 KiB
C++
978 lines
37 KiB
C++
/////////////////////////////////////////////////////////////////////////////////
|
|
//
|
|
// Copyright (C) 2025- Equinor ASA
|
|
//
|
|
// ResInsight is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// ResInsight 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 General Public License at <http://www.gnu.org/licenses/gpl.html>
|
|
// for more details.
|
|
//
|
|
/////////////////////////////////////////////////////////////////////////////////
|
|
|
|
#include "RiaOpenTelemetryManager.h"
|
|
|
|
#include "RiaLogging.h"
|
|
#include "RiaPreferencesOpenTelemetry.h"
|
|
#include "RiaQStringFormatter.h"
|
|
#include "RifJsonEncodeDecode.h"
|
|
|
|
#include <QCryptographicHash>
|
|
#include <QEventLoop>
|
|
#include <QJsonArray>
|
|
#include <QJsonDocument>
|
|
#include <QJsonObject>
|
|
#include <QNetworkAccessManager>
|
|
#include <QNetworkReply>
|
|
#include <QNetworkRequest>
|
|
#include <QProcessEnvironment>
|
|
#include <QRegularExpression>
|
|
#include <QString>
|
|
#include <QSysInfo>
|
|
#include <QTimer>
|
|
#include <QVariantList>
|
|
|
|
#include <algorithm>
|
|
#include <random>
|
|
#include <sstream>
|
|
#include <thread>
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
double RiaOpenTelemetryManager::HealthSnapshot::getSuccessRate() const
|
|
{
|
|
uint64_t total = eventsSent + eventsDropped;
|
|
if ( total == 0 ) return 1.0;
|
|
return static_cast<double>( eventsSent ) / total;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::HealthSnapshot::isHealthy() const
|
|
{
|
|
// Consider healthy if success rate > 90% and we've had recent successful sends
|
|
const auto now = std::chrono::steady_clock::now();
|
|
const auto timeSinceLastSuccess = now - lastSuccessfulSend;
|
|
|
|
return getSuccessRate() > 0.9 && timeSinceLastSuccess < std::chrono::minutes( 5 );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
RiaOpenTelemetryManager& RiaOpenTelemetryManager::instance()
|
|
{
|
|
static RiaOpenTelemetryManager instance;
|
|
return instance;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
/// Get system username in a cross-platform way
|
|
/// Returns username from environment variables (USERNAME on Windows, USER on Linux)
|
|
//--------------------------------------------------------------------------------------------------
|
|
QString RiaOpenTelemetryManager::getSystemUsername()
|
|
{
|
|
QProcessEnvironment env = QProcessEnvironment::systemEnvironment();
|
|
|
|
// Try Windows environment variable first
|
|
QString username = env.value( "USERNAME" );
|
|
if ( !username.isEmpty() )
|
|
{
|
|
return username;
|
|
}
|
|
|
|
// Try Linux/Unix environment variable. Can be empty on some systems
|
|
return env.value( "USER" );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
RiaOpenTelemetryManager::RiaOpenTelemetryManager()
|
|
: QObject( nullptr )
|
|
{
|
|
m_healthMetrics.systemStartTime = std::chrono::steady_clock::now();
|
|
m_healthMetrics.lastSuccessfulSend = m_healthMetrics.systemStartTime;
|
|
|
|
// Initialize Qt networking components
|
|
m_networkAccessManager = new QNetworkAccessManager( this );
|
|
|
|
// Initialize timers
|
|
m_processTimer = new QTimer( this );
|
|
connect( m_processTimer, &QTimer::timeout, this, &RiaOpenTelemetryManager::onProcessEventTimer );
|
|
|
|
m_healthTimer = new QTimer( this );
|
|
connect( m_healthTimer, &QTimer::timeout, this, &RiaOpenTelemetryManager::sendHealthSpan );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
RiaOpenTelemetryManager::~RiaOpenTelemetryManager()
|
|
{
|
|
shutdown();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::initialize()
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
|
|
// Skip initialization if no config file was found (connection string is empty)
|
|
auto* prefs = RiaPreferencesOpenTelemetry::current();
|
|
if ( !prefs || prefs->connectionString().isEmpty() )
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if ( m_initialized.load() )
|
|
{
|
|
return true;
|
|
}
|
|
|
|
if ( !initializeProvider() )
|
|
{
|
|
return false;
|
|
}
|
|
|
|
// Start event processing timer (100ms interval for responsive processing)
|
|
m_isShuttingDown = false;
|
|
m_processTimer->start( 100 );
|
|
|
|
// Start health monitoring timer (5 minutes interval)
|
|
if ( m_healthMonitoringEnabled )
|
|
{
|
|
m_healthTimer->start( 5 * 60 * 1000 ); // 5 minutes in milliseconds
|
|
sendHealthSpan();
|
|
}
|
|
|
|
m_initialized = true;
|
|
m_enabled = true;
|
|
|
|
RiaLogging::info( "OpenTelemetry initialized successfully" );
|
|
|
|
return true;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::shutdown()
|
|
{
|
|
if ( !m_initialized.load() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
m_isShuttingDown = true;
|
|
m_enabled = false;
|
|
|
|
// Stop timers
|
|
if ( m_processTimer )
|
|
{
|
|
m_processTimer->stop();
|
|
}
|
|
if ( m_healthTimer )
|
|
{
|
|
m_healthTimer->stop();
|
|
}
|
|
|
|
// Disconnect network manager to prevent callbacks during shutdown
|
|
if ( m_networkAccessManager )
|
|
{
|
|
m_networkAccessManager->disconnect();
|
|
}
|
|
|
|
// Flush pending events
|
|
flushPendingEvents();
|
|
|
|
m_initialized = false;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::reinitialize()
|
|
{
|
|
shutdown();
|
|
return initialize();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::isEventAllowed( const std::string& eventName ) const
|
|
{
|
|
// Crash events are always reported regardless of filter settings
|
|
if ( eventName.starts_with( "crash." ) ) return true;
|
|
|
|
auto* prefs = RiaPreferencesOpenTelemetry::current();
|
|
if ( !prefs ) return true;
|
|
|
|
const QString name = QString::fromStdString( eventName );
|
|
|
|
auto matches = []( const QString& name, const QStringList& patterns ) -> bool
|
|
{
|
|
for ( const QString& pattern : patterns )
|
|
{
|
|
QRegularExpression re( QRegularExpression::wildcardToRegularExpression( pattern.trimmed() ) );
|
|
if ( re.match( name ).hasMatch() ) return true;
|
|
}
|
|
return false;
|
|
};
|
|
|
|
const QStringList allowlist = prefs->eventAllowlist();
|
|
if ( !allowlist.isEmpty() && !matches( name, allowlist ) ) return false;
|
|
|
|
const QStringList denylist = prefs->eventDenylist();
|
|
return denylist.isEmpty() || !matches( name, denylist );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
namespace
|
|
{
|
|
// Create a stable, anonymized hash of username for privacy-preserving telemetry
|
|
// Uses SHA-256 to ensure consistent hashing across sessions while protecting user identity
|
|
std::string hashUsername( const std::string& username )
|
|
{
|
|
if ( username.empty() )
|
|
{
|
|
return "";
|
|
}
|
|
|
|
QByteArray usernameBytes = QString::fromStdString( username ).toUtf8();
|
|
QByteArray hash = QCryptographicHash::hash( usernameBytes, QCryptographicHash::Sha256 );
|
|
|
|
// Convert to hex string for readability in telemetry
|
|
return hash.toHex().toStdString();
|
|
}
|
|
|
|
void addOsInfo( std::map<std::string, std::string>& attributes )
|
|
{
|
|
attributes["os.type"] = QSysInfo::productType().toStdString();
|
|
attributes["os.version"] = QSysInfo::productVersion().toStdString();
|
|
attributes["os.name"] = QSysInfo::prettyProductName().toStdString();
|
|
}
|
|
} // namespace
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::reportEventAsync( const std::string& eventName, const std::map<std::string, std::string>& attributes )
|
|
{
|
|
const bool isCrashEvent = eventName.starts_with( "crash." );
|
|
|
|
if ( !isCrashEvent && ( !isEnabled() || isCircuitBreakerOpen() ) )
|
|
{
|
|
return;
|
|
}
|
|
|
|
if ( !isEventAllowed( eventName ) )
|
|
{
|
|
return;
|
|
}
|
|
|
|
std::unique_lock<std::mutex> lock( m_queueMutex );
|
|
|
|
// Check queue size and apply backpressure (crash events always bypass the queue limit)
|
|
if ( !isCrashEvent && m_backpressureEnabled && m_eventQueue.size() >= m_maxQueueSize )
|
|
{
|
|
m_healthMetrics.eventsDropped++;
|
|
return;
|
|
}
|
|
|
|
// Create mutable copy and add hashed username if configured (privacy-preserving for regular events)
|
|
// Note: crash events already have hashed username added in reportCrash()
|
|
std::map<std::string, std::string> enrichedAttributes = attributes;
|
|
|
|
if ( !isCrashEvent )
|
|
{
|
|
std::lock_guard<std::mutex> configLock( m_configMutex );
|
|
if ( !m_userHash.empty() )
|
|
{
|
|
enrichedAttributes["user.hash"] = m_userHash;
|
|
}
|
|
|
|
addOsInfo( enrichedAttributes );
|
|
}
|
|
|
|
m_eventQueue.emplace( eventName, enrichedAttributes );
|
|
m_healthMetrics.eventsQueued++;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
/// Helper function to extract the relevant part of the file path starting from "ResInsight"
|
|
/// Returns empty string if "ResInsight" is not found in the path
|
|
//--------------------------------------------------------------------------------------------------
|
|
static std::string extractRelevantPath( const std::string& fullPath )
|
|
{
|
|
size_t pos = fullPath.find( "ResInsight" );
|
|
if ( pos != std::string::npos )
|
|
{
|
|
return fullPath.substr( pos );
|
|
}
|
|
return fullPath;
|
|
}
|
|
|
|
#if RIA_HAS_STD_STACKTRACE
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::reportCrash( int signalCode, const std::stacktrace& trace, const std::map<std::string, std::string>& extraAttributes )
|
|
{
|
|
if ( !isEnabled() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
// Format stack trace as string for logging
|
|
std::stringstream ss;
|
|
int frame = 0;
|
|
for ( const auto& entry : trace )
|
|
{
|
|
std::string relevantPath = extractRelevantPath( entry.source_file() );
|
|
ss << " [" << frame++ << "] " << entry.description() << " at " << relevantPath << ":" << entry.source_line() << "\n";
|
|
}
|
|
std::string rawStackTrace = ss.str();
|
|
|
|
// Store structured stack trace data for Application Insights parsedStack
|
|
QJsonArray jsonFrames;
|
|
int frameIndex = 0;
|
|
for ( const auto& entry : trace )
|
|
{
|
|
std::string relevantPath = extractRelevantPath( entry.source_file() );
|
|
std::string methodName = entry.description();
|
|
|
|
// Ensure method name is never empty (Application Insights requirement)
|
|
if ( methodName.empty() )
|
|
{
|
|
methodName = "UnknownMethod";
|
|
}
|
|
|
|
int lineNumber = static_cast<int>( entry.source_line() );
|
|
|
|
QJsonObject frame;
|
|
frame["level"] = frameIndex;
|
|
frame["method"] = QString::fromStdString( methodName );
|
|
frame["fileName"] = QString::fromStdString( relevantPath );
|
|
frame["line"] = lineNumber;
|
|
|
|
jsonFrames.append( frame );
|
|
frameIndex++;
|
|
}
|
|
|
|
std::map<std::string, std::string> attributes;
|
|
attributes["crash.signal"] = std::to_string( signalCode );
|
|
attributes["crash.thread_id"] = std::to_string( std::hash<std::thread::id>{}( std::this_thread::get_id() ) );
|
|
attributes["crash.stack_trace"] = rawStackTrace;
|
|
attributes["crash.parsed_stack_json"] = QString::fromUtf8( QJsonDocument( jsonFrames ).toJson( QJsonDocument::Compact ) ).toStdString();
|
|
attributes["service.name"] = RiaPreferencesOpenTelemetry::current()->serviceName().toStdString();
|
|
attributes["service.version"] = RiaPreferencesOpenTelemetry::current()->serviceVersion().toStdString();
|
|
|
|
// Merge platform-specific crash context (fault address, signal code, program counter)
|
|
for ( const auto& [key, value] : extraAttributes )
|
|
{
|
|
attributes[key] = value;
|
|
}
|
|
|
|
// Add system information
|
|
addOsInfo( attributes );
|
|
|
|
// Add hashed username so crashes can be grouped per user without revealing the cleartext name.
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
if ( !m_userHash.empty() )
|
|
{
|
|
attributes["user.hash"] = m_userHash;
|
|
}
|
|
}
|
|
|
|
// Send the crash event as its own request instead of appending it to the event queue.
|
|
// flushPendingEvents() drains a single batch from the front of the queue, so a crash event queued behind
|
|
// more than maxBatchSize events would not be part of the request the wait loop below waits for.
|
|
sendBatch( { Event( "crash.signal_handler", attributes ) } );
|
|
|
|
// Flush whatever regular telemetry is still queued. Issued after the crash request so the crash report
|
|
// gets the connection first.
|
|
flushPendingEvents();
|
|
|
|
// Wait for the pending network reply to complete before the process exits.
|
|
// The crash report is sent as an async HTTP POST; without pumping the event loop here
|
|
// the reply never gets a chance to finish and the crash is silently dropped.
|
|
//
|
|
// This runs from within the crash/signal handler, so exclude user input events while pumping:
|
|
// dispatching buffered mouse/keyboard events here would re-enter arbitrary UI code mid-crash and
|
|
// can re-trigger the faulting path (a double-fault inside the handler). Socket and timer events
|
|
// still flow, so the HTTP reply completes.
|
|
if ( m_pendingReplies.load() > 0 )
|
|
{
|
|
auto* prefs = RiaPreferencesOpenTelemetry::current();
|
|
int timeoutMs = prefs ? prefs->connectionTimeoutMs() : 5000;
|
|
int elapsedMs = 0;
|
|
int pollIntervalMs = 50;
|
|
while ( m_pendingReplies.load() > 0 && elapsedMs < timeoutMs )
|
|
{
|
|
QEventLoop loop;
|
|
QTimer::singleShot( pollIntervalMs, &loop, &QEventLoop::quit );
|
|
loop.exec( QEventLoop::ExcludeUserInputEvents );
|
|
elapsedMs += pollIntervalMs;
|
|
}
|
|
}
|
|
|
|
RiaLogging::error( std::format( "Crash reported to OpenTelemetry (signal: {})", signalCode ) );
|
|
}
|
|
#endif // RIA_HAS_STD_STACKTRACE
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::isEnabled() const
|
|
{
|
|
return m_enabled.load() && m_initialized.load();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::isInitialized() const
|
|
{
|
|
return m_initialized.load();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::setErrorCallback( ErrorCallback callback )
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
m_errorCallback = callback;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::setUsername( const std::string& username )
|
|
{
|
|
// Hash at the boundary so the cleartext username is never retained in memory.
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
m_userHash = hashUsername( username );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::setMaxQueueSize( size_t maxEvents )
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
m_maxQueueSize = maxEvents;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::enableBackpressure( bool enable )
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
m_backpressureEnabled = enable;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::setMemoryThreshold( size_t maxMemoryMB )
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
m_memoryThresholdMB = maxMemoryMB;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
size_t RiaOpenTelemetryManager::getCurrentQueueSize() const
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_queueMutex );
|
|
return m_eventQueue.size();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
RiaOpenTelemetryManager::HealthSnapshot RiaOpenTelemetryManager::getHealthMetrics() const
|
|
{
|
|
HealthSnapshot result;
|
|
result.eventsQueued = m_healthMetrics.eventsQueued.load();
|
|
result.eventsSent = m_healthMetrics.eventsSent.load();
|
|
result.eventsDropped = m_healthMetrics.eventsDropped.load();
|
|
result.networkFailures = m_healthMetrics.networkFailures.load();
|
|
result.lastSuccessfulSend = m_healthMetrics.lastSuccessfulSend;
|
|
result.systemStartTime = m_healthMetrics.systemStartTime;
|
|
return result;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::isHealthy() const
|
|
{
|
|
return getHealthMetrics().isHealthy() && !isCircuitBreakerOpen();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::enableHealthMonitoring( bool enable )
|
|
{
|
|
std::lock_guard<std::mutex> lock( m_configMutex );
|
|
m_healthMonitoringEnabled = enable;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::initializeProvider()
|
|
{
|
|
try
|
|
{
|
|
if ( !createExporter() )
|
|
{
|
|
return false;
|
|
}
|
|
|
|
setupResourceAttributes();
|
|
return true;
|
|
}
|
|
catch ( const std::exception& e )
|
|
{
|
|
handleError( TelemetryError::InternalError, QString( "Failed to initialize provider: %1" ).arg( e.what() ) );
|
|
return false;
|
|
}
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
/// Parse Azure Application Insights connection string
|
|
/// Format: InstrumentationKey=<key>;IngestionEndpoint=<endpoint>;...
|
|
//--------------------------------------------------------------------------------------------------
|
|
static std::map<QString, QString> parseAzureConnectionString( const QString& connectionString )
|
|
{
|
|
std::map<QString, QString> result;
|
|
QStringList parts = connectionString.split( ';', Qt::SkipEmptyParts );
|
|
|
|
for ( const QString& part : parts )
|
|
{
|
|
int equalPos = part.indexOf( '=' );
|
|
if ( equalPos > 0 )
|
|
{
|
|
QString key = part.left( equalPos ).trimmed();
|
|
QString value = part.mid( equalPos + 1 ).trimmed();
|
|
result[key] = value;
|
|
}
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::createExporter()
|
|
{
|
|
try
|
|
{
|
|
auto* prefs = RiaPreferencesOpenTelemetry::current();
|
|
if ( !prefs )
|
|
{
|
|
handleError( TelemetryError::ConfigurationError, "Failed to get OpenTelemetry Preferences" );
|
|
return false;
|
|
}
|
|
|
|
// Parse and validate connection string
|
|
auto connectionParams = parseAzureConnectionString( prefs->connectionString() );
|
|
if ( !connectionParams.contains( "InstrumentationKey" ) )
|
|
{
|
|
handleError( TelemetryError::ConfigurationError, "InstrumentationKey not found in connection string" );
|
|
return false;
|
|
}
|
|
|
|
if ( !connectionParams.contains( "IngestionEndpoint" ) )
|
|
{
|
|
handleError( TelemetryError::ConfigurationError, "IngestionEndpoint not found in connection string" );
|
|
return false;
|
|
}
|
|
|
|
// Using Application Insights REST API for telemetry
|
|
RiaLogging::info( "Application Insights REST API configured for production environment" );
|
|
|
|
return true;
|
|
}
|
|
catch ( const std::exception& e )
|
|
{
|
|
handleError( TelemetryError::NetworkError, QString( "Failed to create exporter: %1" ).arg( e.what() ) );
|
|
return false;
|
|
}
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::setupResourceAttributes()
|
|
{
|
|
// Resource attributes are typically set during provider creation
|
|
// This would be expanded with system information, process details, etc.
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::onProcessEventTimer()
|
|
{
|
|
if ( m_isShuttingDown.load() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
if ( isCircuitBreakerOpen() )
|
|
{
|
|
// Do not drain the queue while the breaker is open. attemptReconnection() is rate limited internally, and
|
|
// closes the breaker so the next tick makes a single probe attempt.
|
|
attemptReconnection();
|
|
return;
|
|
}
|
|
|
|
processEvents();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::processEvents( bool ignoreCircuitBreaker )
|
|
{
|
|
if ( !ignoreCircuitBreaker )
|
|
{
|
|
if ( isCircuitBreakerOpen() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
// Keep at most one request in flight. Telemetry failures are only observed when the reply finishes, which for
|
|
// an unreachable endpoint takes the full transfer timeout. Without this, the 100 ms timer keeps launching
|
|
// requests during that window, and they end up aborting each other by saturating the connection pool.
|
|
if ( m_pendingReplies.load() > 0 )
|
|
{
|
|
return;
|
|
}
|
|
}
|
|
|
|
std::unique_lock<std::mutex> lock( m_queueMutex );
|
|
|
|
if ( m_eventQueue.empty() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
// Collect a batch of events
|
|
auto* prefs = RiaPreferencesOpenTelemetry::current();
|
|
int maxBatchSize = prefs ? prefs->maxBatchSize() : 100;
|
|
|
|
std::vector<Event> batch;
|
|
batch.reserve( std::min<size_t>( static_cast<size_t>( std::max( maxBatchSize, 0 ) ), m_eventQueue.size() ) );
|
|
|
|
for ( int i = 0; i < maxBatchSize && !m_eventQueue.empty(); ++i )
|
|
{
|
|
batch.push_back( m_eventQueue.front() );
|
|
m_eventQueue.pop();
|
|
}
|
|
|
|
lock.unlock();
|
|
|
|
// Send events outside of lock
|
|
sendBatch( batch );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::sendBatch( const std::vector<Event>& events )
|
|
{
|
|
if ( events.empty() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
auto* prefs = RiaPreferencesOpenTelemetry::current();
|
|
if ( !prefs )
|
|
{
|
|
handleError( TelemetryError::InternalError, QString( "Failed to access Open Telemetry Preferences." ) );
|
|
updateHealthMetrics( false );
|
|
return;
|
|
}
|
|
|
|
// Use Application Insights REST API
|
|
auto connectionParams = parseAzureConnectionString( prefs->connectionString() );
|
|
|
|
if ( !connectionParams.contains( "InstrumentationKey" ) || !connectionParams.contains( "IngestionEndpoint" ) )
|
|
{
|
|
updateHealthMetrics( false );
|
|
return;
|
|
}
|
|
|
|
// Application Insights accepts an array of envelopes in a single request. Sending one request per event
|
|
// saturates the connection pool of QNetworkAccessManager, and the queued requests are then aborted by their
|
|
// own transfer timeout before they ever reach the wire.
|
|
QVariantList envelope;
|
|
envelope.reserve( static_cast<qsizetype>( events.size() ) );
|
|
|
|
for ( const Event& event : events )
|
|
{
|
|
// Format timestamp - must match Application Insights format exactly
|
|
const auto timestampUtc = std::chrono::floor<std::chrono::milliseconds>( event.timestamp );
|
|
const std::string timestamp = std::format( "{:%Y-%m-%dT%H:%M:%SZ}", timestampUtc );
|
|
|
|
// Convert attributes to JSON properties
|
|
QMap<QString, QVariant> properties;
|
|
for ( const auto& [key, value] : event.attributes )
|
|
{
|
|
properties[QString::fromStdString( key )] = QString::fromStdString( value );
|
|
}
|
|
|
|
// Determine if this is a crash event
|
|
bool isCrashEvent = ( event.name == "crash.signal_handler" );
|
|
|
|
QMap<QString, QVariant> baseData;
|
|
QMap<QString, QVariant> data;
|
|
QMap<QString, QVariant> telemetryItem;
|
|
|
|
if ( isCrashEvent )
|
|
{
|
|
// Create ExceptionData for crash reports
|
|
QMap<QString, QVariant> exception;
|
|
exception["typeName"] = QString( "ResInsightCrash" );
|
|
exception["message"] = QString( "Application crash (signal: %1)" ).arg( properties["crash.signal"].toString() );
|
|
exception["hasFullStack"] = true;
|
|
exception["stack"] = properties["crash.stack_trace"];
|
|
|
|
// Parse structured stack trace for Application Insights
|
|
QString parsedStackJson = properties["crash.parsed_stack_json"].toString();
|
|
if ( !parsedStackJson.isEmpty() )
|
|
{
|
|
// Parse the JSON array of stack frames
|
|
QJsonDocument doc = QJsonDocument::fromJson( parsedStackJson.toUtf8() );
|
|
if ( doc.isArray() )
|
|
{
|
|
exception["parsedStack"] = doc.array().toVariantList();
|
|
}
|
|
}
|
|
|
|
QList<QVariant> exceptions;
|
|
exceptions.append( exception );
|
|
|
|
// Remove stack trace data from properties as it's now in the exception
|
|
properties.remove( "crash.stack_trace" );
|
|
properties.remove( "crash.parsed_stack_json" );
|
|
|
|
baseData["ver"] = 2;
|
|
baseData["exceptions"] = exceptions;
|
|
baseData["properties"] = properties;
|
|
|
|
data["baseType"] = "ExceptionData";
|
|
data["baseData"] = baseData;
|
|
|
|
telemetryItem["name"] = "Microsoft.ApplicationInsights.Exception";
|
|
}
|
|
else
|
|
{
|
|
// Create EventData for regular events
|
|
baseData["ver"] = 2;
|
|
baseData["name"] = QString::fromStdString( event.name );
|
|
baseData["properties"] = properties;
|
|
|
|
data["baseType"] = "EventData";
|
|
data["baseData"] = baseData;
|
|
|
|
telemetryItem["name"] = "Microsoft.ApplicationInsights.Event";
|
|
}
|
|
|
|
telemetryItem["time"] = QString::fromStdString( timestamp );
|
|
telemetryItem["iKey"] = connectionParams["InstrumentationKey"];
|
|
telemetryItem["data"] = data;
|
|
|
|
envelope.append( telemetryItem );
|
|
}
|
|
|
|
// Convert to JSON string
|
|
QString jsonPayload = ResInsightInternalJson::Json::encode( envelope, false );
|
|
|
|
// Send to Application Insights using QNetworkAccessManager
|
|
QString url = connectionParams["IngestionEndpoint"] + "v2/track";
|
|
|
|
QNetworkRequest request;
|
|
request.setUrl( QUrl( url ) );
|
|
request.setHeader( QNetworkRequest::ContentTypeHeader, "application/json" );
|
|
request.setHeader( QNetworkRequest::KnownHeaders( QNetworkRequest::UserAgentHeader ), "ResInsight-OpenTelemetry" );
|
|
request.setTransferTimeout( prefs->connectionTimeoutMs() );
|
|
|
|
const int eventCount = static_cast<int>( events.size() );
|
|
|
|
m_pendingReplies++;
|
|
QNetworkReply* reply = m_networkAccessManager->post( request, jsonPayload.toUtf8() );
|
|
|
|
// Handle response asynchronously
|
|
connect( reply,
|
|
&QNetworkReply::finished,
|
|
this,
|
|
[this, reply, eventCount]()
|
|
{
|
|
const int statusCode = reply->attribute( QNetworkRequest::HttpStatusCodeAttribute ).toInt();
|
|
|
|
// An aborted reply (transfer timeout, shutdown) has a closed device. Reading from it emits
|
|
// "QIODevice::read (QNetworkReplyHttpImpl): device not open" for every such reply.
|
|
const QString responseBody = reply->isOpen() ? QString::fromUtf8( reply->readAll() ) : QString();
|
|
|
|
if ( reply->error() == QNetworkReply::NoError )
|
|
{
|
|
updateHealthMetrics( true );
|
|
resetCircuitBreaker();
|
|
}
|
|
else
|
|
{
|
|
const QString errorMsg =
|
|
QString( "HTTP %1: %2 (%3)" ).arg( statusCode ).arg( reply->errorString() ).arg( responseBody );
|
|
handleError( TelemetryError::NetworkError,
|
|
QString( "Failed to send telemetry (%1 events): %2" ).arg( eventCount ).arg( errorMsg ) );
|
|
updateHealthMetrics( false );
|
|
}
|
|
m_pendingReplies--;
|
|
reply->deleteLater();
|
|
} );
|
|
}
|
|
catch ( const std::exception& e )
|
|
{
|
|
handleError( TelemetryError::InternalError, QString( "Failed to process events: %1" ).arg( e.what() ) );
|
|
updateHealthMetrics( false );
|
|
}
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::flushPendingEvents()
|
|
{
|
|
// Process remaining events in the queue. This is the crash and shutdown path, so bypass the circuit breaker:
|
|
// a crash report must be attempted even when regular telemetry has been suspended.
|
|
processEvents( true );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::handleError( TelemetryError error, const QString& context )
|
|
{
|
|
const int failureCount = m_consecutiveFailures.fetch_add( 1 ) + 1;
|
|
|
|
if ( failureCount >= 3 && !m_circuitBreakerOpen.exchange( true ) )
|
|
{
|
|
// Start the backoff window here. m_lastReconnectAttempt is default constructed to the steady_clock
|
|
// epoch, so without this the first attemptReconnection() call passes the 5 minute check immediately
|
|
// and closes the breaker again on the next timer tick.
|
|
m_lastReconnectAttempt = std::chrono::steady_clock::now();
|
|
|
|
// Only log the transition. Logging on every failure floods the message panel, as the failures that open the
|
|
// breaker keep arriving after it is open.
|
|
RiaLogging::warning( "OpenTelemetry circuit breaker opened due to consecutive failures. Telemetry is "
|
|
"suspended until the connection recovers." );
|
|
}
|
|
|
|
if ( m_errorCallback )
|
|
{
|
|
m_errorCallback( error, context );
|
|
}
|
|
|
|
RiaLogging::warning( std::format( "OpenTelemetry error: {}", context ) );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::attemptReconnection()
|
|
{
|
|
auto now = std::chrono::steady_clock::now();
|
|
if ( now - m_lastReconnectAttempt < std::chrono::minutes( 5 ) )
|
|
{
|
|
return; // Don't retry too frequently
|
|
}
|
|
|
|
m_lastReconnectAttempt = now;
|
|
|
|
// Close the breaker so the next batch is attempted. If the endpoint is still unreachable the breaker reopens
|
|
// after a few failures, and the next probe is another 5 minutes out.
|
|
resetCircuitBreaker();
|
|
RiaLogging::debug( "OpenTelemetry circuit breaker reset, retrying telemetry" );
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
bool RiaOpenTelemetryManager::isCircuitBreakerOpen() const
|
|
{
|
|
return m_circuitBreakerOpen.load();
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::resetCircuitBreaker()
|
|
{
|
|
m_consecutiveFailures = 0;
|
|
m_circuitBreakerOpen = false;
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::updateHealthMetrics( bool success )
|
|
{
|
|
if ( success )
|
|
{
|
|
m_healthMetrics.eventsSent++;
|
|
m_healthMetrics.lastSuccessfulSend = std::chrono::steady_clock::now();
|
|
}
|
|
else
|
|
{
|
|
m_healthMetrics.networkFailures++;
|
|
}
|
|
}
|
|
|
|
//--------------------------------------------------------------------------------------------------
|
|
///
|
|
//--------------------------------------------------------------------------------------------------
|
|
void RiaOpenTelemetryManager::sendHealthSpan()
|
|
{
|
|
if ( !isEnabled() )
|
|
{
|
|
return;
|
|
}
|
|
|
|
auto metrics = getHealthMetrics();
|
|
std::map<std::string, std::string> attributes;
|
|
attributes["health.events_queued"] = std::to_string( metrics.eventsQueued );
|
|
attributes["health.events_sent"] = std::to_string( metrics.eventsSent );
|
|
attributes["health.events_dropped"] = std::to_string( metrics.eventsDropped );
|
|
attributes["health.network_failures"] = std::to_string( metrics.networkFailures );
|
|
attributes["health.success_rate"] = std::to_string( metrics.getSuccessRate() );
|
|
attributes["health.queue_size"] = std::to_string( getCurrentQueueSize() );
|
|
|
|
reportEventAsync( "health.status", attributes );
|
|
} |