/* * dbMango * * Copyright 2025 Deutsche Bank AG * SPDX-License-Identifier: Apache-2.0 * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ using System.Collections.Concurrent; using System.Net.Sockets; using System.Text; namespace Rms.Service.Bootstrap.Logging; internal class SplunkMessageSender(Func getOptions, ILogger? errorHandler) { private const int TimeoutSec = 3; private readonly BlockingCollection _jobs = new(); private readonly CancellationTokenSource _cancelToken = new(); private TcpClient? _client; private Thread? _jobThread; private DateTime _connectionTime; private readonly Lock _syncObj = new (); public void Log(string message) { try { var opt = getOptions(); if ( opt.RemotePort <= 0 || string.IsNullOrWhiteSpace(opt.RemoteHost) ) return; if (_jobThread == null) { _jobThread = new(ThreadProc) { IsBackground = true, Name = "TcpAppender" }; _jobThread.Start(); } //if the queue gets to the limit (100k) then this will drop the event if (_jobs.Count > 1000) errorHandler?.LogError("Job queue reached capacity dropping event"); else _jobs.Add(message); } catch (Exception ex) { errorHandler?.LogError($"Unable to send logging event to {SplunkAddress}", ex); } } private string SplunkAddress => string.IsNullOrWhiteSpace(getOptions().RemoteHost) || getOptions().RemotePort <= 0 ? "" : $"tcp://{getOptions().RemoteHost}:{getOptions().RemotePort}"; /// /// Thread proc which loops sending messages to the endpoint /// private void ThreadProc() { try { while (!_cancelToken.IsCancellationRequested) { //will block until an event is available var loggingEvent = _jobs.Take(_cancelToken.Token); while (!CheckConnection() && !_cancelToken.IsCancellationRequested) { //keep spinning until we can reconnect, with 5 mins between attempts Thread.Sleep((int)TimeSpan.FromMinutes(1).TotalMilliseconds); } if (!_cancelToken.IsCancellationRequested) SendLoggingEvent(loggingEvent); } } catch (OperationCanceledException /*e*/) { //thrown when jobs.take is cancelled, do nothing, just shut down } catch (Exception ex) { errorHandler?.LogError($"Exception in thread proc {ex.Message}"); } } /// /// Send a message via the tcp connection /// /// /// private void SendLoggingEvent(string loggingData) { try { lock (_syncObj) { if (_client is not { Connected: true }) return; } var buffer = Encoding.UTF8.GetBytes(loggingData); lock (_syncObj) { _client.GetStream().Write(buffer, 0, buffer.Length); } } catch (Exception ex) { errorHandler?.LogError($"Unable to send logging event to {SplunkAddress}", ex); } } /// /// Encure the connection to the endpoint is connected /// Will retry 5 times if there's a failure /// /// private bool CheckConnection() { lock (_syncObj) { if (_client is not { Connected: true } || DateTime.UtcNow - _connectionTime > TimeSpan.FromMinutes(1)) { _client = new() { SendTimeout = (int)TimeSpan.FromSeconds(TimeoutSec).TotalMilliseconds }; } //try to connect to the end point, will retry 5 times, with a backoff timeout between each attempt //2,4,6,8,10 seconds between attempts var retries = 0; while (!_client.Connected && !_cancelToken.IsCancellationRequested && retries++ < 5) { if ( getOptions().RemotePort <= 0 || string.IsNullOrWhiteSpace(getOptions().RemoteHost)) { Thread.Sleep((int)TimeSpan.FromSeconds(1).TotalMilliseconds); continue; } try { var result = _client.BeginConnect(getOptions().RemoteHost, getOptions().RemotePort, null, null); result.AsyncWaitHandle.WaitOne(TimeSpan.FromSeconds(TimeoutSec)); // we have connected _client.EndConnect(result); //remember the time we connected, we will regularly reconnect to make sure splunk doesnt keeping dropping us. _connectionTime = DateTime.UtcNow; } catch (Exception ex) { Thread.Sleep((int)TimeSpan.FromSeconds(2 * retries).TotalMilliseconds); errorHandler?.LogError($"Unable to connect to {SplunkAddress} attempt={retries}", ex); } } if (!_client.Connected) { errorHandler?.LogError($"Failed all attempts to connect to {SplunkAddress}"); return false; } return true; } } }