1+ using System ;
2+ using System . Collections . Generic ;
3+ using System . Threading . Tasks ;
4+ using AWS . Lambda . Powertools . Metrics ;
5+ using Xunit ;
6+
7+ namespace AWS . Lambda . Powertools . Metrics . Tests
8+ {
9+ public class ConcurrencyIssueTest : IDisposable
10+ {
11+ [ Fact ]
12+ public async Task AddMetric_ConcurrentAccess_ShouldNotThrowException ( )
13+ {
14+ // Arrange
15+ Metrics . ResetForTest ( ) ;
16+ Metrics . SetNamespace ( "TestNamespace" ) ;
17+ var exceptions = new List < Exception > ( ) ;
18+ var tasks = new List < Task > ( ) ;
19+
20+ // Act - Simulate concurrent access from multiple threads
21+ for ( int i = 0 ; i < 10 ; i ++ )
22+ {
23+ var taskId = i ;
24+ tasks . Add ( Task . Run ( ( ) =>
25+ {
26+ try
27+ {
28+ // Simulate multiple metrics being added concurrently
29+ for ( int j = 0 ; j < 100 ; j ++ )
30+ {
31+ Metrics . AddMetric ( $ "TestMetric_{ taskId } _{ j } ", 1.0 , MetricUnit . Count ) ;
32+ Metrics . AddMetric ( $ "Client.{ taskId } ", 1.0 , MetricUnit . Count ) ;
33+ Metrics . AddMetadata ( $ "TestMetadata_{ taskId } _{ j } ", $ "value_{ j } ") ;
34+ }
35+ }
36+ catch ( Exception ex )
37+ {
38+ lock ( exceptions )
39+ {
40+ exceptions . Add ( ex ) ;
41+ }
42+ }
43+ } ) ) ;
44+ }
45+
46+ await Task . WhenAll ( tasks ) ;
47+
48+ // Assert
49+ foreach ( var ex in exceptions )
50+ {
51+ Console . WriteLine ( $ "Exception: { ex . GetType ( ) . Name } : { ex . Message } ") ;
52+ if ( ex . StackTrace != null )
53+ Console . WriteLine ( $ "Stack trace: { ex . StackTrace } ") ;
54+ }
55+ Assert . Empty ( exceptions ) ;
56+
57+ // Cleanup after test
58+ CleanupMetrics ( ) ;
59+ }
60+
61+ [ Fact ]
62+ public async Task AddMetric_ConcurrentAccessWithSameKey_ShouldNotThrowException ( )
63+ {
64+ // Arrange
65+ Metrics . ResetForTest ( ) ;
66+ Metrics . SetNamespace ( "TestNamespace" ) ;
67+ var exceptions = new List < Exception > ( ) ;
68+ var tasks = new List < Task > ( ) ;
69+
70+ // Act - Simulate the specific scenario where the same metric key is used concurrently
71+ // Increase concurrency to try to reproduce the issue
72+ for ( int i = 0 ; i < 50 ; i ++ )
73+ {
74+ tasks . Add ( Task . Run ( ( ) =>
75+ {
76+ try
77+ {
78+ // This simulates the scenario where the same metric key
79+ // (like "Client.6b70*28198e") is being added from multiple threads
80+ for ( int j = 0 ; j < 200 ; j ++ )
81+ {
82+ Metrics . AddMetric ( "Client.SharedKey" , 1.0 , MetricUnit . Count ) ;
83+ }
84+ }
85+ catch ( Exception ex )
86+ {
87+ lock ( exceptions )
88+ {
89+ exceptions . Add ( ex ) ;
90+ }
91+ }
92+ } ) ) ;
93+ }
94+
95+ await Task . WhenAll ( tasks ) ;
96+
97+ // Assert - Should not have any exceptions
98+ foreach ( var ex in exceptions )
99+ {
100+ Console . WriteLine ( $ "Exception: { ex . GetType ( ) . Name } : { ex . Message } ") ;
101+ if ( ex . StackTrace != null )
102+ Console . WriteLine ( $ "Stack trace: { ex . StackTrace } ") ;
103+ }
104+ Assert . Empty ( exceptions ) ;
105+
106+ // Cleanup after test
107+ CleanupMetrics ( ) ;
108+ }
109+
110+ [ Fact ]
111+ public async Task AddMetric_Batch_ShouldNotThrowException ( )
112+ {
113+ // Arrange
114+ Metrics . ResetForTest ( ) ;
115+ Metrics . SetNamespace ( "TestNamespace" ) ;
116+ var exceptions = new List < Exception > ( ) ;
117+ var tasks = new List < Task > ( ) ;
118+
119+ for ( int i = 0 ; i < 5 ; i ++ )
120+ {
121+ var batchId = i ;
122+ tasks . Add ( Task . Run ( async ( ) =>
123+ {
124+ try
125+ {
126+ // Simulate DataLoader batch processing
127+ var innerTasks = new List < Task > ( ) ;
128+ for ( int j = 0 ; j < 10 ; j ++ )
129+ {
130+ var itemId = j ;
131+ innerTasks . Add ( Task . Run ( ( ) =>
132+ {
133+ // Simulate metrics being added from parallel DataLoader operations
134+ Metrics . AddMetric ( $ "DataLoader.InsidersStatusDataLoader", 1.0 , MetricUnit . Count ) ;
135+ Metrics . AddMetric ( $ "Query.insidersStatus", 1.0 , MetricUnit . Count ) ;
136+ Metrics . AddMetric ( $ "Client.6b70*28198e", 1.0 , MetricUnit . Count ) ;
137+ Metrics . AddMetadata ( $ "Query.insidersStatus.OperationName", "GetInsidersStatus" ) ;
138+ Metrics . AddMetadata ( $ "Query.insidersStatus.UserId", $ "user_{ batchId } _{ itemId } ") ;
139+ } ) ) ;
140+ }
141+ await Task . WhenAll ( innerTasks ) ;
142+ }
143+ catch ( Exception ex )
144+ {
145+ lock ( exceptions )
146+ {
147+ exceptions . Add ( ex ) ;
148+ }
149+ }
150+ } ) ) ;
151+ }
152+
153+ await Task . WhenAll ( tasks ) ;
154+
155+ // Assert
156+ foreach ( var ex in exceptions )
157+ {
158+ Console . WriteLine ( $ "Exception: { ex . GetType ( ) . Name } : { ex . Message } ") ;
159+ if ( ex . StackTrace != null )
160+ Console . WriteLine ( $ "Stack trace: { ex . StackTrace } ") ;
161+ }
162+ Assert . Empty ( exceptions ) ;
163+
164+ // Cleanup after test
165+ CleanupMetrics ( ) ;
166+ }
167+
168+ [ Fact ]
169+ public async Task AddMetric_ReproduceFirstOrDefaultIssue_ShouldNotThrowException ( )
170+ {
171+ // Arrange
172+ Metrics . ResetForTest ( ) ;
173+ Metrics . SetNamespace ( "TestNamespace" ) ;
174+ var exceptions = new List < Exception > ( ) ;
175+ var tasks = new List < Task > ( ) ;
176+
177+ // Act - This test specifically targets the FirstOrDefault issue in line 202-203 of Metrics.cs
178+ // metrics are added and flushed rapidly to trigger collection modification
179+ for ( int i = 0 ; i < 100 ; i ++ )
180+ {
181+ var taskId = i ;
182+ tasks . Add ( Task . Run ( ( ) =>
183+ {
184+ try
185+ {
186+ // Add metrics rapidly to trigger the overflow condition that calls FirstOrDefault
187+ for ( int j = 0 ; j < 150 ; j ++ ) // This should trigger multiple flushes
188+ {
189+ Metrics . AddMetric ( $ "TestMetric_{ taskId } _{ j } ", 1.0 , MetricUnit . Count ) ;
190+
191+ // Also add the same metric key to trigger the FirstOrDefault path
192+ if ( j % 10 == 0 )
193+ {
194+ Metrics . AddMetric ( "SharedMetric" , 1.0 , MetricUnit . Count ) ;
195+ }
196+ }
197+ }
198+ catch ( Exception ex )
199+ {
200+ lock ( exceptions )
201+ {
202+ exceptions . Add ( ex ) ;
203+ }
204+ }
205+ } ) ) ;
206+ }
207+
208+ await Task . WhenAll ( tasks ) ;
209+
210+ // Assert
211+ foreach ( var ex in exceptions )
212+ {
213+ Console . WriteLine ( $ "Exception: { ex . GetType ( ) . Name } : { ex . Message } ") ;
214+ if ( ex . StackTrace != null )
215+ Console . WriteLine ( $ "Stack trace: { ex . StackTrace } ") ;
216+ }
217+ Assert . Empty ( exceptions ) ;
218+
219+ // Cleanup after test
220+ CleanupMetrics ( ) ;
221+ }
222+
223+ /// <summary>
224+ /// Cleanup method to ensure no state leaks between tests
225+ /// </summary>
226+ private void CleanupMetrics ( )
227+ {
228+ try
229+ {
230+ // Reset the static instance to clean state
231+ Metrics . ResetForTest ( ) ;
232+ }
233+ catch
234+ {
235+ // Ignore cleanup errors
236+ }
237+ }
238+
239+ /// <summary>
240+ /// IDisposable implementation for proper test cleanup
241+ /// </summary>
242+ public void Dispose ( )
243+ {
244+ CleanupMetrics ( ) ;
245+ }
246+ }
247+ }
0 commit comments