Add BizTalk endpoint reachability monitoring
This commit is contained in:
@@ -0,0 +1,383 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace BizTalkCheckmkPulse
|
||||
{
|
||||
/// <summary>
|
||||
/// Pflegt den woechentlichen Endpoint-Katalog und prueft aktive Ziele parallel auf Netzwerkebene.
|
||||
/// </summary>
|
||||
internal sealed class EndpointConnectivityProbe
|
||||
{
|
||||
private readonly MonitoringOptions _options;
|
||||
private readonly FileLogger _logger;
|
||||
|
||||
public EndpointConnectivityProbe(MonitoringOptions options, FileLogger logger)
|
||||
{
|
||||
_options = options;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public void Query(ProbeResult result)
|
||||
{
|
||||
var state = result.EndpointConnectivity;
|
||||
if (!_options.ProbeEndpointConnectivity)
|
||||
{
|
||||
state.Disabled = true;
|
||||
return;
|
||||
}
|
||||
|
||||
state.RuntimeStateAvailable = result.Platform.ReceiveLocationsDataAvailable
|
||||
&& result.Platform.SendPortsDataAvailable;
|
||||
if (!state.RuntimeStateAvailable)
|
||||
{
|
||||
state.Failure = "Aktive Send Ports und Receive Locations konnten nicht vollstaendig bestimmt werden.";
|
||||
return;
|
||||
}
|
||||
|
||||
var store = new EndpointCatalogStore(
|
||||
_options.EndpointCatalogPath,
|
||||
_options.EndpointCatalogMaxBytes,
|
||||
_options.EndpointMaxCount);
|
||||
EndpointCatalog catalog = null;
|
||||
string readFailure = null;
|
||||
try
|
||||
{
|
||||
catalog = store.Read(_options.EnvironmentName);
|
||||
}
|
||||
catch (FileNotFoundException ex)
|
||||
{
|
||||
readFailure = ex.Message;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
readFailure = ex.GetType().Name + ": " + ex.Message;
|
||||
_logger.Warning("Endpoint catalog rejected. path=" + _options.EndpointCatalogPath + " reason=" + readFailure);
|
||||
}
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
state.RefreshRequired = catalog == null
|
||||
|| catalog.SynchronizedUtc > now.AddMinutes(5)
|
||||
|| now - catalog.SynchronizedUtc >= TimeSpan.FromHours(_options.EndpointDiscoveryIntervalHours);
|
||||
if (state.RefreshRequired)
|
||||
{
|
||||
try
|
||||
{
|
||||
catalog = Synchronize(catalog, result.EndpointCandidates, now);
|
||||
store.Write(catalog);
|
||||
state.RefreshSucceeded = true;
|
||||
_logger.Info(
|
||||
"Endpoint catalog synchronized. path=" + _options.EndpointCatalogPath
|
||||
+ " active_candidates=" + catalog.ActiveCandidates
|
||||
+ " configured=" + catalog.Entries.Count
|
||||
+ " unsupported_external=" + catalog.UnsupportedCandidates);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
state.RefreshSucceeded = false;
|
||||
state.Failure = "Endpoint-Katalog konnte nicht synchronisiert werden: " + ex.GetType().Name + ": " + ex.Message;
|
||||
_logger.Error("Endpoint catalog synchronization failed.", ex);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
state.RefreshSucceeded = true;
|
||||
}
|
||||
|
||||
if (catalog == null)
|
||||
{
|
||||
state.CatalogAvailable = false;
|
||||
if (string.IsNullOrWhiteSpace(state.Failure))
|
||||
{
|
||||
state.Failure = "Keine gueltige Endpoint-Konfiguration verfuegbar: " + readFailure;
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
state.CatalogAvailable = true;
|
||||
state.CatalogSynchronizedUtc = catalog.SynchronizedUtc;
|
||||
state.Configured = catalog.Entries.Count;
|
||||
state.UnsupportedActive = CountUnsupportedExternal(result.EndpointCandidates);
|
||||
|
||||
var activeKeys = new HashSet<string>(
|
||||
result.EndpointCandidates.Where(x => x.Active).Select(x => x.Key),
|
||||
StringComparer.OrdinalIgnoreCase);
|
||||
var activeEntries = catalog.Entries
|
||||
.Where(x => x.Enabled)
|
||||
.Where(x => string.Equals(x.ArtifactType, "Manual", StringComparison.OrdinalIgnoreCase)
|
||||
|| activeKeys.Contains(x.Key))
|
||||
.Take(_options.EndpointMaxCount)
|
||||
.ToArray();
|
||||
state.Active = activeEntries.Length;
|
||||
state.SkippedInactive = catalog.Entries.Count(x => x.Enabled) - activeEntries.Length;
|
||||
|
||||
try
|
||||
{
|
||||
foreach (var probeResult in ProbeAllAsync(activeEntries).GetAwaiter().GetResult())
|
||||
{
|
||||
state.Results.Add(probeResult);
|
||||
if (!probeResult.Available)
|
||||
{
|
||||
_logger.Warning(
|
||||
"Endpoint unavailable. endpoint=" + Display(probeResult.Endpoint)
|
||||
+ " duration_ms=" + probeResult.DurationMilliseconds
|
||||
+ " reason=" + probeResult.Failure);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
state.Failure = "Endpoint-Probes konnten nicht abgeschlossen werden: " + ex.GetType().Name + ": " + ex.Message;
|
||||
_logger.Error("Endpoint probes failed.", ex);
|
||||
}
|
||||
}
|
||||
|
||||
internal static EndpointCatalog SynchronizeCatalog(
|
||||
EndpointCatalog existing,
|
||||
IEnumerable<EndpointCandidate> candidates,
|
||||
DateTime synchronizedUtc)
|
||||
{
|
||||
var active = (candidates ?? Enumerable.Empty<EndpointCandidate>())
|
||||
.Where(x => x != null && x.Active)
|
||||
.ToArray();
|
||||
var discovered = new List<EndpointCatalogEntry>();
|
||||
var unsupported = 0;
|
||||
foreach (var candidate in active)
|
||||
{
|
||||
EndpointCatalogEntry entry;
|
||||
string reason;
|
||||
if (EndpointAddressParser.TryCreate(candidate, out entry, out reason))
|
||||
{
|
||||
discovered.Add(entry);
|
||||
}
|
||||
else if (EndpointAddressParser.IsPotentialExternalEndpoint(candidate))
|
||||
{
|
||||
unsupported++;
|
||||
}
|
||||
}
|
||||
|
||||
var manual = existing == null
|
||||
? new EndpointCatalogEntry[0]
|
||||
: existing.Entries.Where(x => !x.AutoDiscovered).ToArray();
|
||||
var manualKeys = new HashSet<string>(manual.Select(x => x.Key), StringComparer.OrdinalIgnoreCase);
|
||||
var merged = manual
|
||||
.Concat(discovered.Where(x => !manualKeys.Contains(x.Key)))
|
||||
.GroupBy(x => x.Key, StringComparer.OrdinalIgnoreCase)
|
||||
.Select(x => x.First())
|
||||
.ToArray();
|
||||
|
||||
var catalog = new EndpointCatalog
|
||||
{
|
||||
MachineName = Environment.MachineName,
|
||||
EnvironmentName = existing == null ? string.Empty : existing.EnvironmentName,
|
||||
SynchronizedUtc = synchronizedUtc.ToUniversalTime(),
|
||||
ActiveCandidates = active.Length,
|
||||
UnsupportedCandidates = unsupported
|
||||
};
|
||||
catalog.Entries.AddRange(merged);
|
||||
return catalog;
|
||||
}
|
||||
|
||||
private EndpointCatalog Synchronize(
|
||||
EndpointCatalog existing,
|
||||
IEnumerable<EndpointCandidate> candidates,
|
||||
DateTime synchronizedUtc)
|
||||
{
|
||||
var catalog = SynchronizeCatalog(existing, candidates, synchronizedUtc);
|
||||
catalog.EnvironmentName = _options.EnvironmentName ?? string.Empty;
|
||||
if (catalog.Entries.Count > _options.EndpointMaxCount
|
||||
|| catalog.ActiveCandidates > _options.EndpointMaxCount)
|
||||
{
|
||||
throw new InvalidDataException("Discovered endpoint candidates exceed EndpointMaxCount.");
|
||||
}
|
||||
|
||||
return catalog;
|
||||
}
|
||||
|
||||
private static int CountUnsupportedExternal(IEnumerable<EndpointCandidate> candidates)
|
||||
{
|
||||
var count = 0;
|
||||
foreach (var candidate in candidates.Where(x => x.Active))
|
||||
{
|
||||
EndpointCatalogEntry ignored;
|
||||
string reason;
|
||||
if (!EndpointAddressParser.TryCreate(candidate, out ignored, out reason)
|
||||
&& EndpointAddressParser.IsPotentialExternalEndpoint(candidate))
|
||||
{
|
||||
count++;
|
||||
}
|
||||
}
|
||||
|
||||
return count;
|
||||
}
|
||||
|
||||
private async Task<IReadOnlyCollection<EndpointProbeResult>> ProbeAllAsync(EndpointCatalogEntry[] endpoints)
|
||||
{
|
||||
var unique = endpoints
|
||||
.GroupBy(x => x.Protocol.ToUpperInvariant() + "|" + x.Host.ToUpperInvariant() + "|" + x.Port)
|
||||
.ToArray();
|
||||
var outcomes = new Dictionary<string, NetworkOutcome>(StringComparer.OrdinalIgnoreCase);
|
||||
using (var gate = new SemaphoreSlim(_options.EndpointProbeMaxConcurrency))
|
||||
{
|
||||
var tasks = unique.Select(async group =>
|
||||
{
|
||||
await gate.WaitAsync().ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
var endpoint = group.First();
|
||||
var outcome = await ProbeOneAsync(endpoint).ConfigureAwait(false);
|
||||
lock (outcomes)
|
||||
{
|
||||
outcomes[group.Key] = outcome;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
gate.Release();
|
||||
}
|
||||
}).ToArray();
|
||||
await Task.WhenAll(tasks).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return endpoints.Select(endpoint =>
|
||||
{
|
||||
var key = endpoint.Protocol.ToUpperInvariant() + "|" + endpoint.Host.ToUpperInvariant() + "|" + endpoint.Port;
|
||||
var outcome = outcomes[key];
|
||||
return new EndpointProbeResult
|
||||
{
|
||||
Endpoint = endpoint,
|
||||
Available = outcome.Available,
|
||||
DurationMilliseconds = outcome.DurationMilliseconds,
|
||||
Failure = outcome.Failure
|
||||
};
|
||||
}).ToArray();
|
||||
}
|
||||
|
||||
private async Task<NetworkOutcome> ProbeOneAsync(EndpointCatalogEntry endpoint)
|
||||
{
|
||||
return string.Equals(endpoint.Protocol, "UDP", StringComparison.OrdinalIgnoreCase)
|
||||
? await ProbeUdpAsync(endpoint).ConfigureAwait(false)
|
||||
: await ProbeTcpAsync(endpoint).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task<NetworkOutcome> ProbeTcpAsync(EndpointCatalogEntry endpoint)
|
||||
{
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
using (var client = new TcpClient())
|
||||
{
|
||||
try
|
||||
{
|
||||
var connect = client.ConnectAsync(endpoint.Host, endpoint.Port);
|
||||
var completed = await Task.WhenAny(connect, Task.Delay(_options.EndpointProbeTimeoutMilliseconds)).ConfigureAwait(false);
|
||||
if (completed != connect)
|
||||
{
|
||||
client.Close();
|
||||
ObserveFault(connect);
|
||||
return NetworkOutcome.Failed(stopwatch.ElapsedMilliseconds, "TCP timeout");
|
||||
}
|
||||
|
||||
await connect.ConfigureAwait(false);
|
||||
return client.Connected
|
||||
? NetworkOutcome.Success(stopwatch.ElapsedMilliseconds)
|
||||
: NetworkOutcome.Failed(stopwatch.ElapsedMilliseconds, "TCP connection not established");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return NetworkOutcome.Failed(stopwatch.ElapsedMilliseconds, CompactFailure(ex));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<NetworkOutcome> ProbeUdpAsync(EndpointCatalogEntry endpoint)
|
||||
{
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
try
|
||||
{
|
||||
var resolution = Dns.GetHostAddressesAsync(endpoint.Host);
|
||||
var completed = await Task.WhenAny(resolution, Task.Delay(_options.EndpointProbeTimeoutMilliseconds)).ConfigureAwait(false);
|
||||
if (completed != resolution)
|
||||
{
|
||||
ObserveFault(resolution);
|
||||
return NetworkOutcome.Failed(stopwatch.ElapsedMilliseconds, "UDP DNS timeout");
|
||||
}
|
||||
|
||||
var addresses = await resolution.ConfigureAwait(false);
|
||||
var address = addresses.FirstOrDefault(x => x.AddressFamily == AddressFamily.InterNetwork)
|
||||
?? addresses.FirstOrDefault();
|
||||
if (address == null)
|
||||
{
|
||||
return NetworkOutcome.Failed(stopwatch.ElapsedMilliseconds, "UDP host resolved without an address");
|
||||
}
|
||||
|
||||
using (var socket = new Socket(address.AddressFamily, SocketType.Dgram, ProtocolType.Udp))
|
||||
{
|
||||
socket.Connect(new IPEndPoint(address, endpoint.Port));
|
||||
socket.Send(new byte[0]);
|
||||
}
|
||||
|
||||
// UDP ist verbindungslos: Erfolg bestaetigt DNS, Route und lokalen Datagrammversand,
|
||||
// aber ohne applikationsspezifische Antwort nicht den entfernten Dienstzustand.
|
||||
return NetworkOutcome.Success(stopwatch.ElapsedMilliseconds);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return NetworkOutcome.Failed(stopwatch.ElapsedMilliseconds, CompactFailure(ex));
|
||||
}
|
||||
}
|
||||
|
||||
private static void ObserveFault(Task task)
|
||||
{
|
||||
task.ContinueWith(
|
||||
completed => { var ignored = completed.Exception; },
|
||||
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously);
|
||||
}
|
||||
|
||||
private static string CompactFailure(Exception exception)
|
||||
{
|
||||
var socket = exception as SocketException;
|
||||
return socket == null
|
||||
? exception.GetType().Name + ": " + exception.Message
|
||||
: "SocketError=" + socket.SocketErrorCode + ": " + socket.Message;
|
||||
}
|
||||
|
||||
internal static string Display(EndpointCatalogEntry endpoint)
|
||||
{
|
||||
var artifact = string.IsNullOrWhiteSpace(endpoint.ApplicationName)
|
||||
|| string.Equals(endpoint.ApplicationName, "(unknown)", StringComparison.OrdinalIgnoreCase)
|
||||
? endpoint.ArtifactName
|
||||
: endpoint.ApplicationName + "\\" + endpoint.ArtifactName;
|
||||
return endpoint.ArtifactType + ":" + artifact
|
||||
+ "[" + endpoint.TransportRole + "]->"
|
||||
+ endpoint.Host + ":" + endpoint.Port + "/" + endpoint.Protocol.ToUpperInvariant();
|
||||
}
|
||||
|
||||
private sealed class NetworkOutcome
|
||||
{
|
||||
public bool Available { get; private set; }
|
||||
public long DurationMilliseconds { get; private set; }
|
||||
public string Failure { get; private set; }
|
||||
|
||||
public static NetworkOutcome Success(long durationMilliseconds)
|
||||
{
|
||||
return new NetworkOutcome { Available = true, DurationMilliseconds = durationMilliseconds };
|
||||
}
|
||||
|
||||
public static NetworkOutcome Failed(long durationMilliseconds, string failure)
|
||||
{
|
||||
return new NetworkOutcome
|
||||
{
|
||||
Available = false,
|
||||
DurationMilliseconds = durationMilliseconds,
|
||||
Failure = failure
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user