1919using Microsoft . Extensions . Hosting ;
2020using Microsoft . Extensions . Logging ;
2121using Microsoft . Rest ;
22+ using Timer = System . Timers . Timer ;
2223
2324namespace ES . Kubernetes . Reflector . CertManager
2425{
@@ -29,22 +30,26 @@ public class Monitor : IHostedService, IHealthCheck
2930 private readonly Dictionary < string , ManagedWatcher < Certificate , object > > _certificatesWatchers =
3031 new Dictionary < string , ManagedWatcher < Certificate , object > > ( ) ;
3132
32- private readonly ManagedWatcher < V1CustomResourceDefinition , V1CustomResourceDefinitionList > _crdWatcher ;
33+ private readonly ManagedWatcher < V1CustomResourceDefinition , V1CustomResourceDefinitionList > _crdV1Watcher ;
34+
3335 private readonly FeederQueue < WatcherEvent > _eventQueue ;
3436 private readonly ILogger < Monitor > _logger ;
3537 private readonly IMediator _mediator ;
3638 private readonly ManagedWatcher < V1Secret , V1SecretList > _secretsWatcher ;
3739 private readonly IKubernetes _apiClient ;
3840
41+ private readonly Timer _v1Beta1CrdMonitorTimer = new Timer ( ) ;
42+ private bool _v1Beta1CrdMonitorTimerFaulted ;
43+
3944 public Monitor ( ILogger < Monitor > logger ,
40- ManagedWatcher < V1CustomResourceDefinition , V1CustomResourceDefinitionList > crdWatcher ,
45+ ManagedWatcher < V1CustomResourceDefinition , V1CustomResourceDefinitionList > crdV1Watcher ,
4146 Func < ManagedWatcher < Certificate , object > > certificatesWatcherFactory ,
4247 ManagedWatcher < V1Secret , V1SecretList > secretsWatcher ,
4348 IKubernetes apiClient ,
4449 IMediator mediator )
4550 {
4651 _logger = logger ;
47- _crdWatcher = crdWatcher ;
52+ _crdV1Watcher = crdV1Watcher ;
4853 _certificatesWatcherFactory = certificatesWatcherFactory ;
4954 _secretsWatcher = secretsWatcher ;
5055 _apiClient = apiClient ;
@@ -65,34 +70,22 @@ public Monitor(ILogger<Monitor> logger,
6570 await c . ListSecretForAllNamespacesWithHttpMessagesAsync ( watch : true ) ;
6671
6772
68- _crdWatcher . EventHandlerFactory = OnCrdEvent ;
69- _crdWatcher . RequestFactory = async c =>
73+ _crdV1Watcher . EventHandlerFactory = OnCrdEventV1 ;
74+ _crdV1Watcher . RequestFactory = async c =>
7075 await c . ListCustomResourceDefinitionWithHttpMessagesAsync ( watch : true ) ;
71- _crdWatcher . OnStateChanged = async ( sender , update ) =>
72- {
73- switch ( update . State )
74- {
75- case ManagedWatcherState . Closed :
76- _logger . LogDebug ( "{type} watcher {state}" , typeof ( V1CustomResourceDefinition ) . Name ,
77- update . State ) ;
78- await sender . Start ( ) ;
79- break ;
80- case ManagedWatcherState . Faulted :
81- _logger . LogError ( update . Exception , "{type} watcher {state}" ,
82- typeof ( V1CustomResourceDefinition ) . Name , update . State ) ;
83- break ;
84- default :
85- _logger . LogDebug ( "{type} watcher {state}" , typeof ( V1CustomResourceDefinition ) . Name ,
86- update . State ) ;
87- break ;
88- }
89- } ;
76+ _crdV1Watcher . OnStateChanged = OnCrdWatcherStateChanged ;
77+
78+ _v1Beta1CrdMonitorTimer . Elapsed += ( _ , __ ) => Onv1Beta1CrdRefresh ( ) ;
79+ _v1Beta1CrdMonitorTimer . Interval = 30_000 ;
9080 }
9181
82+
9283 public Task < HealthCheckResult > CheckHealthAsync ( HealthCheckContext context ,
9384 CancellationToken cancellationToken = new CancellationToken ( ) )
9485 {
95- return Task . FromResult ( _crdWatcher . IsFaulted || _secretsWatcher . IsFaulted ||
86+ return Task . FromResult ( _crdV1Watcher . IsFaulted ||
87+ _v1Beta1CrdMonitorTimerFaulted ||
88+ _secretsWatcher . IsFaulted ||
9689 _certificatesWatchers . Values . Any ( s => s . IsFaulted )
9790 ? HealthCheckResult . Unhealthy ( )
9891 : HealthCheckResult . Healthy ( ) ) ;
@@ -103,31 +96,94 @@ public async Task StartAsync(CancellationToken cancellationToken)
10396 try
10497 {
10598 await _apiClient . ListCustomResourceDefinitionAsync ( cancellationToken : cancellationToken ) ;
106- await _crdWatcher . Start ( ) ;
99+ await _crdV1Watcher . Start ( ) ;
107100 }
108101 catch ( HttpOperationException exception ) when ( exception . Response . StatusCode == HttpStatusCode . NotFound )
109102 {
110- _logger . LogError (
103+ _logger . LogWarning (
111104 "Current kubernetes version does not support {type} apiVersion {version}." ,
112105 V1CustomResourceDefinition . KubeKind , V1CustomResourceDefinition . KubeApiVersion ) ;
106+ Onv1Beta1CrdRefresh ( ) ;
107+ _v1Beta1CrdMonitorTimer . Start ( ) ;
113108 }
114109 }
115110
116111 public async Task StopAsync ( CancellationToken cancellationToken )
117112 {
118- await _crdWatcher . Stop ( ) ;
113+ await _crdV1Watcher . Stop ( ) ;
119114 foreach ( var certificatesWatcher in _certificatesWatchers . Values ) await certificatesWatcher . Stop ( ) ;
120115 await _secretsWatcher . Stop ( ) ;
121116 }
122117
123- private async Task OnWatcherStateChanged < TS , TSL > ( ManagedWatcher < TS , TSL , WatcherEvent < TS > > sender ,
124- ManagedWatcherStateUpdate update ) where TS : class , IKubernetesObject
118+ private void Onv1Beta1CrdRefresh ( )
119+ {
120+ try
121+ {
122+ _logger . LogDebug (
123+ "Updating {type} {kind} in group {group}" ,
124+ typeof ( V1beta1CustomResourceDefinition ) . Name ,
125+ CertManagerConstants . CertificateKind , CertManagerConstants . CrdGroup ) ;
126+
127+ _v1Beta1CrdMonitorTimerFaulted = false ;
128+ var crd = _apiClient . ListCustomResourceDefinition1 ( ) . Items
129+ . FirstOrDefault ( s =>
130+ s . Spec ? . Names != null && s . Spec . Group == CertManagerConstants . CrdGroup &&
131+ s . Spec . Names . Kind == CertManagerConstants . CertificateKind ) ;
132+
133+ if ( crd == null ) return ;
134+ OnCrdVersionUpdate ( crd . GetType ( ) . Name , crd . Spec . Names . Kind ,
135+ crd . Spec . Group , crd . Spec . Names . Plural , crd . Spec . Versions . Select ( s => s . Name ) . ToList ( ) ) . Wait ( ) ;
136+
137+
138+ }
139+ catch ( HttpOperationException exception ) when ( exception . Response . StatusCode == HttpStatusCode . NotFound )
140+ {
141+ _logger . LogWarning (
142+ "Current kubernetes version does not support {type} apiVersion {version}." ,
143+ V1beta1CustomResourceDefinition . KubeKind , V1beta1CustomResourceDefinition . KubeApiVersion ) ;
144+
145+ _v1Beta1CrdMonitorTimer . Stop ( ) ;
146+ }
147+ catch ( Exception exception )
148+ {
149+ _v1Beta1CrdMonitorTimerFaulted = true ;
150+ _v1Beta1CrdMonitorTimer . Stop ( ) ;
151+ _logger . LogError ( exception ,
152+ "Error occured while getting {kind} version {version}" ,
153+ V1beta1CustomResourceDefinition . KubeKind , V1beta1CustomResourceDefinition . KubeApiVersion ) ;
154+ }
155+
156+
157+ }
158+
159+ private async Task OnCrdWatcherStateChanged < TResource , TResourceList > (
160+ ManagedWatcher < TResource , TResourceList , WatcherEvent < TResource > > sender , ManagedWatcherStateUpdate update )
161+ where TResource : class , IKubernetesObject
162+ {
163+ switch ( update . State )
164+ {
165+ case ManagedWatcherState . Closed :
166+ _logger . LogDebug ( "{type} watcher {state}" , typeof ( TResource ) . Name , update . State ) ;
167+ await sender . Start ( ) ;
168+ break ;
169+ case ManagedWatcherState . Faulted :
170+ _logger . LogError ( update . Exception , "{type} watcher {state}" ,
171+ typeof ( TResource ) . Name , update . State ) ;
172+ break ;
173+ default :
174+ _logger . LogDebug ( "{type} watcher {state}" , typeof ( TResource ) . Name , update . State ) ;
175+ break ;
176+ }
177+ }
178+
179+ private async Task OnWatcherStateChanged < TResource , TResourceList > ( ManagedWatcher < TResource , TResourceList , WatcherEvent < TResource > > sender ,
180+ ManagedWatcherStateUpdate update ) where TResource : class , IKubernetesObject
125181 {
126182 var tag = sender . Tag ?? string . Empty ;
127183 switch ( update . State )
128184 {
129185 case ManagedWatcherState . Closed :
130- _logger . LogDebug ( "{type} watcher {tag} {state}" , typeof ( TS ) . Name , tag , update . State ) ;
186+ _logger . LogDebug ( "{type} watcher {tag} {state}" , typeof ( TResource ) . Name , tag , update . State ) ;
131187 await _secretsWatcher . Stop ( ) ;
132188 foreach ( var certificatesWatcher in _certificatesWatchers . Values ) await certificatesWatcher . Stop ( ) ;
133189
@@ -137,11 +193,11 @@ private async Task OnWatcherStateChanged<TS, TSL>(ManagedWatcher<TS, TSL, Watche
137193 foreach ( var certificatesWatcher in _certificatesWatchers . Values ) await certificatesWatcher . Start ( ) ;
138194 break ;
139195 case ManagedWatcherState . Faulted :
140- _logger . LogError ( update . Exception , "{type} watcher {tag} {state}" , typeof ( TS ) . Name , tag ,
196+ _logger . LogError ( update . Exception , "{type} watcher {tag} {state}" , typeof ( TResource ) . Name , tag ,
141197 update . State ) ;
142198 break ;
143199 default :
144- _logger . LogDebug ( "{type} watcher {tag} {state}" , typeof ( TS ) . Name , tag , update . State ) ;
200+ _logger . LogDebug ( "{type} watcher {tag} {state}" , typeof ( TResource ) . Name , tag , update . State ) ;
145201 break ;
146202 }
147203 }
@@ -169,7 +225,7 @@ private async Task OnEventHandlingError(WatcherEvent e, Exception ex)
169225 }
170226
171227
172- private async Task OnCrdEvent ( WatcherEvent < V1CustomResourceDefinition > request )
228+ private async Task OnCrdEventV1 ( WatcherEvent < V1CustomResourceDefinition > request )
173229 {
174230 if ( request . Type != WatchEventType . Added && request . Type != WatchEventType . Modified ) return ;
175231 if ( request . Item . Spec ? . Names == null ) return ;
@@ -179,11 +235,16 @@ private async Task OnCrdEvent(WatcherEvent<V1CustomResourceDefinition> request)
179235 var versions = request . Item . Spec . Versions . Select ( s => s . Name ) . ToList ( ) ;
180236 if ( versions . TrueForAll ( s => _certificatesWatchers . ContainsKey ( s ) ) ) return ;
181237
238+ await OnCrdVersionUpdate ( request . Item . GetType ( ) . Name , request . Item . Spec . Names . Kind , request . Item . Spec . Group ,
239+ request . Item . Spec . Names . Plural , versions ) ;
240+ }
241+
242+ private async Task OnCrdVersionUpdate ( string crdType , string crdKind , string crdGroup , string crdPlural , List < string > versions )
243+ {
244+ if ( versions . TrueForAll ( s => _certificatesWatchers . ContainsKey ( s ) ) ) return ;
245+
182246 _logger . LogInformation ( "{crdType} {kind} in group {group} versions updated to {versions}" ,
183- request . Item . GetType ( ) . Name ,
184- request . Item . Spec . Names . Kind ,
185- request . Item . Spec . Group ,
186- versions ) ;
247+ crdType , crdKind , crdGroup , versions ) ;
187248
188249 foreach ( var certificatesWatcher in _certificatesWatchers . Values ) await certificatesWatcher . Stop ( ) ;
189250 await _secretsWatcher . Stop ( ) ;
@@ -198,8 +259,7 @@ private async Task OnCrdEvent(WatcherEvent<V1CustomResourceDefinition> request)
198259 watcher . EventHandlerFactory = e =>
199260 _eventQueue . FeedAsync ( new InternalCertificateWatcherEvent { Item = e . Item , Type = e . Type } ) ;
200261 watcher . RequestFactory = async client => await client . ListClusterCustomObjectWithHttpMessagesAsync (
201- request . Item . Spec . Group ,
202- version , request . Item . Spec . Names . Plural , watch : true ,
262+ crdGroup , version , crdPlural , watch : true ,
203263 timeoutSeconds : ( int ) TimeSpan . FromHours ( 1 ) . TotalSeconds ) ;
204264 _certificatesWatchers . Add ( version , watcher ) ;
205265 }
0 commit comments