Skip to content

Commit 4a51fad

Browse files
zbalkanzbalkan
committed
Major refactor
1 parent 37c1016 commit 4a51fad

11 files changed

Lines changed: 170 additions & 158 deletions

File tree

‎Apps/LogExporterApp/App.cs‎

Lines changed: 25 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ You should have received a copy of the GNU General Public License
1818
*/
1919

2020
using DnsServerCore.ApplicationCommon;
21-
using LogExporter.Enrichment;
21+
using LogExporter.Pipeline;
2222
using LogExporter.Sinks;
2323
using System;
2424
using System.Collections.Generic;
@@ -37,7 +37,7 @@ public sealed class App : IDnsApplication, IDnsQueryLogger
3737
private const int BULK_INSERT_COUNT = 1000;
3838

3939
private readonly SinkDispatcher _sinkDispatcher;
40-
private readonly EnrichmentDispatcher _enrichmentDispatcher;
40+
private readonly PipelineDispatcher _enrichmentDispatcher;
4141

4242
// Stage 1 buffer: transformed LogEntry waiting for enrichment
4343
private Channel<LogEntry> _transformChannel = default!;
@@ -62,7 +62,7 @@ public sealed class App : IDnsApplication, IDnsQueryLogger
6262
public App()
6363
{
6464
_sinkDispatcher = new SinkDispatcher();
65-
_enrichmentDispatcher = new EnrichmentDispatcher();
65+
_enrichmentDispatcher = new PipelineDispatcher();
6666
_lastDropTicks = DateTime.UtcNow.Ticks;
6767
}
6868

@@ -150,8 +150,9 @@ public Task InitializeAsync(IDnsServer dnsServer, string config)
150150
_config = AppConfig.Deserialize(config)
151151
?? throw new DnsClientException("Invalid application configuration.");
152152

153+
ConfigurePipeline();
154+
153155
ConfigureSinks();
154-
ConfigureEnrichments();
155156
}
156157
catch (Exception ex)
157158
{
@@ -170,7 +171,7 @@ public Task InitializeAsync(IDnsServer dnsServer, string config)
170171

171172
// Stage 1: transform buffer – InsertLogAsync pushes LogEntry here.
172173
_transformChannel = Channel.CreateBounded<LogEntry>(
173-
new BoundedChannelOptions(_config!.MaxQueueSize)
174+
new BoundedChannelOptions(_config!.Sinks.MaxQueueSize)
174175
{
175176
SingleReader = true,
176177
SingleWriter = false, // InsertLogAsync may be called concurrently
@@ -179,7 +180,7 @@ public Task InitializeAsync(IDnsServer dnsServer, string config)
179180

180181
// Stage 2: enriched buffer – EnrichLogsAsync pushes here, ExportLogsAsync consumes.
181182
_enrichedChannel = Channel.CreateBounded<LogEntry>(
182-
new BoundedChannelOptions(_config.MaxQueueSize)
183+
new BoundedChannelOptions(_config.Sinks.MaxQueueSize)
183184
{
184185
SingleReader = true,
185186
SingleWriter = true, // only enrichment stage writes
@@ -215,7 +216,7 @@ public Task InsertLogAsync(DateTime timestamp, DnsDatagram request,
215216
try
216217
{
217218
// input -> transform: build LogEntry
218-
entry = new LogEntry(timestamp, remoteEP, protocol, request, response, _config!.EnableEdnsLogging);
219+
entry = new LogEntry(timestamp, remoteEP, protocol, request, response, _config!.Sinks.EnableEdnsLogging);
219220
}
220221
catch (Exception ex)
221222
{
@@ -258,7 +259,7 @@ private async Task EnrichLogsAsync()
258259
{
259260
try
260261
{
261-
_enrichmentDispatcher.Enrich(entry, ex => _dnsServer?.WriteLog(ex));
262+
_enrichmentDispatcher.Run(entry, ex => _dnsServer?.WriteLog(ex));
262263
}
263264
catch (Exception ex)
264265
{
@@ -299,7 +300,7 @@ private async Task EnrichLogsAsync()
299300
}
300301
}
301302

302-
// Step 3: ExportLogsAsync – enrich -> output
303+
// Step 3: ExportLogsAsync – pipeline -> output
303304
private async Task ExportLogsAsync()
304305
{
305306
// ADR: Reuse this list buffer to avoid GC churn during high-volume logging.
@@ -343,39 +344,39 @@ await _sinkDispatcher
343344

344345
private void ConfigureSinks()
345346
{
347+
var sinks = _config!.Sinks;
346348
_sinkDispatcher.Remove(typeof(ConsoleSink));
347-
if (_config!.ConsoleSinkConfig != null && _config.ConsoleSinkConfig.Enabled)
349+
if (sinks.ConsoleSinkConfig != null && sinks.ConsoleSinkConfig.Enabled)
348350
_sinkDispatcher.Add(new ConsoleSink());
349351

350352
_sinkDispatcher.Remove(typeof(FileSink));
351-
if (_config.FileSinkConfig?.Enabled is true)
352-
_sinkDispatcher.Add(new FileSink(_config.FileSinkConfig.Path));
353+
if (sinks.FileSinkConfig?.Enabled is true)
354+
_sinkDispatcher.Add(new FileSink(sinks.FileSinkConfig.Path));
353355

354356
_sinkDispatcher.Remove(typeof(HttpSink));
355-
if (_config.HttpSinkConfig?.Enabled is true)
357+
if (sinks.HttpSinkConfig?.Enabled is true)
356358
{
357359
_sinkDispatcher.Add(
358-
new HttpSink(_config.HttpSinkConfig.Endpoint, _config.HttpSinkConfig.Headers));
360+
new HttpSink(sinks.HttpSinkConfig.Endpoint, sinks.HttpSinkConfig.Headers));
359361
}
360362

361363
_sinkDispatcher.Remove(typeof(SyslogSink));
362-
if (_config.SyslogSinkConfig?.Enabled is true)
364+
if (sinks.SyslogSinkConfig?.Enabled is true)
363365
{
364366
_sinkDispatcher.Add(
365-
new SyslogSink(_config.SyslogSinkConfig.Address,
366-
_config.SyslogSinkConfig.Port!.Value,
367-
_config.SyslogSinkConfig.Protocol));
367+
new SyslogSink(sinks.SyslogSinkConfig.Address,
368+
sinks.SyslogSinkConfig.Port!.Value,
369+
sinks.SyslogSinkConfig.Protocol));
368370
}
369371
}
370372

371-
private void ConfigureEnrichments()
373+
private void ConfigurePipeline()
372374
{
373375
// Remove any existing enricher types first to avoid duplicate registration.
374-
_enrichmentDispatcher.Remove(typeof(PublicSuffixEnrichment));
375-
376-
if (_config!.PSLEnrichment?.Enabled is true)
376+
_enrichmentDispatcher.Remove(typeof(Normalize));
377+
if (_config!.Pipeline.NormalizeProcessConfig?.Enabled is true)
377378
{
378-
_enrichmentDispatcher.Add(new PublicSuffixEnrichment());
379+
_enrichmentDispatcher.Add(new Normalize());
379380
}
380381
}
381382

@@ -400,7 +401,7 @@ private void IncrementDropAndMaybeLog()
400401
#region properties
401402

402403
public string Description =>
403-
"Allows exporting query logs to third party sinks. Supports exporting to File, HTTP endpoint, and Syslog.";
404+
"Allows exporting query logs to third party sinks. Supports exporting to FileSink, HTTP endpoint, and SyslogSink.";
404405

405406
#endregion properties
406407
}

‎Apps/LogExporterApp/AppConfig.cs‎

Lines changed: 74 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -18,38 +18,22 @@ You should have received a copy of the GNU General Public License
1818
1919
*/
2020

21-
using System;
2221
using System.Collections.Generic;
2322
using System.ComponentModel.DataAnnotations;
2423
using System.Text.Json;
2524
using System.Text.Json.Serialization;
2625
using TechnitiumLibrary.Net.Dns;
26+
using static LogExporter.SinkConfig;
2727

2828
namespace LogExporter
2929
{
3030
public class AppConfig
3131
{
32-
[JsonPropertyName("maxQueueSize")]
33-
[Range(1, int.MaxValue, ErrorMessage = "maxQueueSize must be greater than zero.")]
34-
public int MaxQueueSize { get; set; }
35-
36-
[JsonPropertyName("enableEdnsLogging")]
37-
public bool EnableEdnsLogging { get; set; }
38-
39-
[JsonPropertyName("enablePslResolution")]
40-
public PSLEnrichment? PSLEnrichment { get; set; }
41-
42-
[JsonPropertyName("console")]
43-
public ConsoleSinkConfig? ConsoleSinkConfig { get; set; }
44-
45-
[JsonPropertyName("file")]
46-
public FileSinkConfig? FileSinkConfig { get; set; }
32+
[JsonPropertyName("sinks")]
33+
public SinkConfig Sinks { get; set; }
4734

48-
[JsonPropertyName("http")]
49-
public HttpSinkConfig? HttpSinkConfig { get; set; }
50-
51-
[JsonPropertyName("syslog")]
52-
public SyslogSinkConfig? SyslogSinkConfig { get; set; }
35+
[JsonPropertyName("pipeline")]
36+
public PipelineConfig Pipeline { get; set; }
5337

5438
/// <summary>
5539
/// Loads config and enforces DataAnnotations validation.
@@ -69,14 +53,15 @@ public static AppConfig Deserialize(string json)
6953
ValidateObject(config);
7054

7155
// Validate enabled targets only — disabled ones may be incomplete by design.
72-
if (config.FileSinkConfig?.Enabled is true)
73-
ValidateObject(config.FileSinkConfig);
7456

75-
if (config.HttpSinkConfig?.Enabled is true)
76-
ValidateObject(config.HttpSinkConfig);
57+
if (config.Sinks.FileSinkConfig?.Enabled is true)
58+
ValidateObject(config.Sinks.FileSinkConfig);
7759

78-
if (config.SyslogSinkConfig?.Enabled is true)
79-
ValidateObject(config.SyslogSinkConfig);
60+
if (config.Sinks.HttpSinkConfig?.Enabled is true)
61+
ValidateObject(config.Sinks.HttpSinkConfig);
62+
63+
if (config.Sinks.SyslogSinkConfig?.Enabled is true)
64+
ValidateObject(config.Sinks.SyslogSinkConfig);
8065

8166
return config;
8267
}
@@ -87,60 +72,83 @@ private static void ValidateObject(object instance)
8772
Validator.ValidateObject(instance, ctx, validateAllProperties: true);
8873
}
8974
}
90-
91-
#region Enrichments
92-
public class EnrichmentConfigBase
75+
public class FeatureBase
9376
{
9477
[JsonPropertyName("enabled")]
95-
public bool Enabled { get; set; }
78+
public bool Enabled { get; set; } = true;
9679
}
9780

98-
public class PSLEnrichment : EnrichmentConfigBase { }
99-
#endregion
100-
101-
#region Sinks
102-
public class SinkConfigBase
81+
public class SinkConfig
10382
{
104-
[JsonPropertyName("enabled")]
105-
public bool Enabled { get; set; }
106-
}
83+
[Range(1, int.MaxValue, ErrorMessage = "maxQueueSize must be greater than zero.")]
10784

108-
public class ConsoleSinkConfig : SinkConfigBase { }
85+
[JsonPropertyName("maxQueueSize")]
86+
public int MaxQueueSize { get; set; } = int.MaxValue;
10987

110-
public class SyslogSinkConfig : SinkConfigBase
111-
{
112-
[JsonPropertyName("address")]
113-
[Required(ErrorMessage = "syslog.address is required when syslog logging is enabled.")]
114-
public string Address { get; set; } = string.Empty;
88+
[JsonPropertyName("enableEdnsLogging")]
89+
public bool EnableEdnsLogging { get; set; } = true;
11590

116-
[JsonPropertyName("port")]
117-
[Range(1, 65535)]
118-
public int? Port { get; set; }
91+
[JsonPropertyName("console")]
92+
public ConsoleSink ConsoleSinkConfig { get; set; }
11993

120-
[JsonPropertyName("protocol")]
121-
[AllowedValues(["UDP", "TCP", "TLS", "LOCAL"])]
122-
public string? Protocol { get; set; }
123-
}
94+
[JsonPropertyName("file")]
95+
public FileSink FileSinkConfig { get; set; }
12496

125-
public class FileSinkConfig : SinkConfigBase
126-
{
127-
[JsonPropertyName("path")]
128-
[Required(ErrorMessage = "file.path is required when file logging is enabled.")]
129-
public string Path { get; set; } = string.Empty;
97+
[JsonPropertyName("http")]
98+
public HttpSink HttpSinkConfig { get; set; }
99+
100+
[JsonPropertyName("syslog")]
101+
public SyslogSink SyslogSinkConfig { get; set; }
102+
103+
public class SyslogSink : FeatureBase
104+
{
105+
[Required(ErrorMessage = "syslog.address is required when syslog logging is enabled.")]
106+
[JsonPropertyName("address")]
107+
public string Address { get; set; }
108+
109+
[Range(1, 65535)]
110+
[JsonPropertyName("port")]
111+
public int? Port { get; set; }
112+
113+
[AllowedValues(["UDP", "TCP", "TLS", "LOCAL"])]
114+
[JsonPropertyName("protocol")]
115+
public string Protocol { get; set; }
116+
}
117+
118+
public class ConsoleSink : FeatureBase
119+
{
120+
}
121+
122+
public class FileSink : FeatureBase
123+
{
124+
[Required(ErrorMessage = "file.path is required when syslog logging is enabled.")]
125+
[JsonPropertyName("path")]
126+
public string Path { get; set; }
127+
}
128+
129+
public class HttpSink : FeatureBase
130+
{
131+
132+
[Required(ErrorMessage = "http.endpoint is required when HTTP logging is enabled.")]
133+
[Url]
134+
[JsonPropertyName("endpoint")]
135+
public string Endpoint { get; set; }
136+
137+
[JsonPropertyName("headers")]
138+
public Dictionary<string, string?>? Headers { get; set; }
139+
}
130140
}
131141

132-
public class HttpSinkConfig : SinkConfigBase
142+
public class PipelineConfig
133143
{
134-
[JsonPropertyName("endpoint")]
135-
[Required(ErrorMessage = "http.endpoint is required when HTTP logging is enabled.")]
136-
[Url]
137-
public string Endpoint { get; set; } = string.Empty;
144+
[JsonPropertyName("normalize")]
145+
public NormalizeProcess NormalizeProcessConfig { get; set; }
138146

139-
[JsonPropertyName("headers")]
140-
public Dictionary<string, string?>? Headers { get; set; }
141-
}
142-
#endregion
147+
public class NormalizeProcess : FeatureBase
148+
{
143149

150+
}
151+
}
144152
/// <summary>
145153
/// Shared serializer configuration for reading dnsApp.config.
146154
/// ADR: The serializer options are centralized so that parsing behavior

‎Apps/LogExporterApp/LogEntry.cs‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ You should have received a copy of the GNU General Public License
1818
1919
*/
2020
using DnsServerCore.ApplicationCommon;
21-
using LogExporter.Enrichment;
2221
using System;
2322
using System.Collections.Generic;
2423
using System.Linq;
@@ -28,7 +27,6 @@ You should have received a copy of the GNU General Public License
2827
using TechnitiumLibrary.Net.Dns;
2928
using TechnitiumLibrary.Net.Dns.EDnsOptions;
3029
using TechnitiumLibrary.Net.Dns.ResourceRecords;
31-
using static LogExporter.Enrichment.PublicSuffixEnrichment;
3230

3331
namespace LogExporter
3432
{
@@ -168,8 +166,8 @@ private void PopulateEDNSLgs(DnsDatagram response, bool ednsLogging)
168166

169167
public DateTime Timestamp { get; }
170168

171-
// Enrichment bag populated by enrichment pipeline stages
172-
public Dictionary<string, object> Enrichment { get; } = new();
169+
// Meta bag populated by pipeline stages
170+
public Dictionary<string, object> Meta { get; } = new();
173171

174172
public override string ToString()
175173
{

Apps/LogExporterApp/Enrichment/IEnrichment.cs renamed to Apps/LogExporterApp/Pipeline/IPipelineProcessor.cs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,10 @@ You should have received a copy of the GNU General Public License
1919

2020
using System;
2121

22-
namespace LogExporter.Enrichment
22+
namespace LogExporter.Pipeline
2323
{
24-
public interface IEnrichment : IDisposable
24+
public interface IPipelineProcessor : IDisposable
2525
{
26-
void Enrich(LogEntry logEntry);
26+
void Process(LogEntry logEntry);
2727
}
2828
}

Apps/LogExporterApp/Enrichment/PublicSuffixEnrichment.DomainCache.cs renamed to Apps/LogExporterApp/Pipeline/Normalize.DomainCache.cs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,10 @@ You should have received a copy of the GNU General Public License
2222
using System;
2323
using System.Collections.Concurrent;
2424

25-
namespace LogExporter.Enrichment
25+
namespace LogExporter.Pipeline
2626
{
2727

28-
public partial class PublicSuffixEnrichment
28+
public partial class Normalize
2929
{
3030
/// <summary>
3131
/// Thread-safe cache for parsed domain information using the SIEVE eviction algorithm.
@@ -70,7 +70,7 @@ public DomainInfo GetOrAdd(string domainName)
7070
return node.Domain;
7171
}
7272

73-
// Normalize only if needed, using string pool to reduce allocations
73+
// NormalizeConfig only if needed, using string pool to reduce allocations
7474
string normalizedName = GetPooledNormalizedName(domainName);
7575

7676
// Check cache again with normalized name (may differ from original)

0 commit comments

Comments
 (0)