11// SPDX-License-Identifier: Apache-2.0
22// SPDX-FileCopyrightText: 2023-Present The Pepr Authors
33
4- import readline from "readline" ;
5-
4+ import byline from "byline" ;
65import fetch from "node-fetch" ;
7- import { GenericClass } from "../types" ;
6+
7+ import { GenericClass , LogFn } from "../types" ;
88import { Filters , WatchAction , WatchPhase } from "./types" ;
99import { k8sCfg , pathBuilder } from "./utils" ;
1010
11+ /**
12+ * Wrapper for the AbortController to allow the watch to be aborted externally.
13+ */
14+ export type WatchController = {
15+ /**
16+ * Abort the watch.
17+ * @param reason optional reason for aborting the watch
18+ * @returns
19+ */
20+ abort : ( reason ?: string ) => void ;
21+ /**
22+ * Get the AbortSignal for the watch.
23+ * @returns
24+ */
25+ signal : ( ) => AbortSignal ;
26+ } ;
27+
28+ /**
29+ * Configuration for the watch function.
30+ */
31+ export type WatchCfg = {
32+ /**
33+ * The maximum number of times to retry the watch, the retry count is reset on success.
34+ */
35+ retryMax ?: number ;
36+ /**
37+ * The delay between retries in seconds.
38+ */
39+ retryDelaySec ?: number ;
40+ /**
41+ * A function to log errors.
42+ */
43+ logFn ?: LogFn ;
44+ /**
45+ * A function to call when the watch fails after the maximum number of retries.
46+ */
47+ retryFail ?: ( e : Error ) => void ;
48+ } ;
49+
1150/**
1251 * Execute a watch on the specified resource.
1352 */
1453export async function ExecWatch < T extends GenericClass > (
1554 model : T ,
1655 filters : Filters ,
1756 callback : WatchAction < T > ,
57+ watchCfg : WatchCfg = { } ,
1858) {
59+ watchCfg . logFn ?.( { model, filters, watchCfg } , "ExecWatch" ) ;
60+
1961 // Build the path and query params for the resource, excluding the name
2062 const { opts, serverUrl } = await k8sCfg ( "GET" ) ;
2163 const url = pathBuilder ( serverUrl , model , filters , true ) ;
@@ -31,64 +73,137 @@ export async function ExecWatch<T extends GenericClass>(
3173 url . searchParams . set ( "fieldSelector" , `metadata.name=${ filters . name } ) ;
3274 }
3375
34- // Add abort controller to the long-running request
35- const controller = new AbortController ( ) ;
36- opts . signal = controller . signal ;
76+ // Set the initial timeout to 15 seconds
77+ opts . timeout = 15 * 1000 ;
78+
79+ // Enable keep alive
80+ ( opts . agent as unknown as { keepAlive : boolean } ) . keepAlive = true ;
81+
82+ // Track the number of retries
83+ let retryCount = 0 ;
84+
85+ // Set the maximum number of retries to 5 if not specified
86+ watchCfg . retryMax ??= 5 ;
87+
88+ // Set the retry delay to 5 seconds if not specified
89+ watchCfg . retryDelaySec ??= 5 ;
90+
91+ // Create a throwaway AbortController to setup the wrapped AbortController
92+ let abortController : AbortController ;
93+
94+ // Create a wrapped AbortController to allow the watch to be aborted externally
95+ const abortWrapper = { } as WatchController ;
96+
97+ function bindAbortController ( ) {
98+ // Create a new AbortController
99+ abortController = new AbortController ( ) ;
100+
101+ // Update the abort wrapper
102+ abortWrapper . abort = reason => abortController . abort ( reason ) ;
103+ abortWrapper . signal = ( ) => abortController . signal ;
104+
105+ // Add the abort signal to the request options
106+ opts . signal = abortController . signal ;
107+ }
108+
109+ async function runner ( ) {
110+ let doneCalled = false ;
37111
38- // Close the connection and make the callback function no-op
39- let close = ( err ?: Error ) => {
40- controller . abort ( ) ;
41- close = ( ) => { } ;
42- if ( err ) {
43- throw err ;
112+ bindAbortController ( ) ;
113+
114+ // Create a stream to read the response body
115+ const stream = byline . createStream ( ) ;
116+
117+ const onError = ( err : Error ) => {
118+ stream . removeAllListeners ( ) ;
119+
120+ if ( ! doneCalled ) {
121+ doneCalled = true ;
122+
123+ // If the error is not an AbortError, reload the watch
124+ if ( err . name !== "AbortError" ) {
125+ watchCfg . logFn ?.( err , "stream error" ) ;
126+ void reload ( err ) ;
127+ } else {
128+ watchCfg . logFn ?.( "watch aborted via WatchController.abort()" ) ;
129+ }
130+ }
131+ } ;
132+
133+ const cleanup = ( ) => {
134+ if ( ! doneCalled ) {
135+ doneCalled = true ;
136+ stream . removeAllListeners ( ) ;
137+ }
138+ } ;
139+
140+ try {
141+ // Make the actual request
142+ const response = await fetch ( url , { ...opts } ) ;
143+
144+ // If the request is successful, start listening for events
145+ if ( response . ok ) {
146+ const { body } = response ;
147+
148+ // Reset the retry count
149+ retryCount = 0 ;
150+
151+ stream . on ( "error" , onError ) ;
152+ stream . on ( "close" , cleanup ) ;
153+ stream . on ( "finish" , cleanup ) ;
154+
155+ // Listen for events and call the callback function
156+ stream . on ( "data" , line => {
157+ try {
158+ // Parse the event payload
159+ const { object : payload , type : phase } = JSON . parse ( line ) as {
160+ type : WatchPhase ;
161+ object : InstanceType < T > ;
162+ } ;
163+
164+ // Call the callback function with the parsed payload
165+ void callback ( payload , phase as WatchPhase ) ;
166+ } catch ( err ) {
167+ watchCfg . logFn ?.( err , "watch callback error" ) ;
168+ }
169+ } ) ;
170+
171+ body . on ( "error" , onError ) ;
172+ body . on ( "close" , cleanup ) ;
173+ body . on ( "finish" , cleanup ) ;
174+
175+ // Pipe the response body to the stream
176+ body . pipe ( stream ) ;
177+ } else {
178+ throw new Error ( `watch failed: ${ response . status } ${ response . statusText } ) ;
179+ }
180+ } catch ( e ) {
181+ onError ( e ) ;
44182 }
45- } ;
46-
47- try {
48- // Make the actual request
49- const response = await fetch ( url , opts ) ;
50-
51- // If the request is successful, start listening for events
52- if ( response . ok ) {
53- const { body } = response ;
54-
55- // Bind connection events to the close function
56- body . on ( "error" , close ) ;
57- body . on ( "close" , close ) ;
58- body . on ( "finish" , close ) ;
59-
60- // Create a readline interface to parse the stream
61- const rl = readline . createInterface ( {
62- input : response . body ! ,
63- terminal : false ,
64- } ) ;
65-
66- // Listen for events and call the callback function
67- rl . on ( "line" , line => {
68- try {
69- // Parse the event payload
70- const { object : payload , type : phase } = JSON . parse ( line ) as {
71- type : WatchPhase ;
72- object : InstanceType < T > ;
73- } ;
74-
75- // Call the callback function with the parsed payload
76- void callback ( payload , phase as WatchPhase ) ;
77- } catch ( ignore ) {
78- // ignore parse errors
183+
184+ // On unhandled errors, retry the watch
185+ async function reload ( e : Error ) {
186+ // If there are more attempts, retry the watch
187+ if ( watchCfg . retryMax ! > retryCount ) {
188+ retryCount ++ ;
189+
190+ watchCfg . logFn ?.( `retrying watch ${ retryCount } ${ watchCfg . retryMax } ) ;
191+
192+ // Sleep for the specified delay or 5 seconds
193+ await new Promise ( r => setTimeout ( r , watchCfg . retryDelaySec ! * 1000 ) ) ;
194+
195+ // Retry the watch after the delay
196+ await runner ( ) ;
197+ } else {
198+ // Otherwise, call the finally function if it exists
199+ if ( watchCfg . retryFail ) {
200+ watchCfg . retryFail ( e ) ;
79201 }
80- } ) ;
81- } else {
82- // If the request fails, throw an error
83- const error = new Error ( response . statusText ) as Error & {
84- statusCode : number | undefined ;
85- } ;
86- error . statusCode = response . status ;
87- throw error ;
202+ }
88203 }
89- } catch ( e ) {
90- close ( e ) ;
91204 }
92205
93- return controller ;
206+ await runner ( ) ;
207+
208+ return abortWrapper ;
94209}
0 commit comments