Repository navigation
Return new WatchEvent for error types on WatchContinuosly instead of an exception. #451
Description
Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
When we WatchContinuously a ConfigMap with:
e.g.:
Source
.single(cm)
.concatLazy {
k8s.watchContinuously[ConfigMap](cm.name, Some(cm.resourceVersion), bufSize = 10 * 1000 * 1000).collect {
case WatchEvent(_type, cm) if _type == EventType.ADDED || _type == EventType.MODIFIED => cm
}
}
When we migrated to Pekko, we receive exceptions like this one:
stack_trace: "skuber.api.client.package$K8SException: {
"apiVersion" : "v1",
"kind" : "Status",
"metadata" : { },
"status" : "Failure",
"message" : "too old resource version: 4440038452 (4441424417)",
"reason" : "Expired",
"code" : 410
}
at skuber.api.watch.BytesToWatchEventSource$.$anonfun$apply$1(BytesToWatchEventSource.scala:21)
at org.apache.pekko.stream.impl.fusing.Map$$anon$1.onPush(Ops.scala:64)
at org.apache.pekko.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:555)
"
And the stream is stopped, because even if we do a .recover, the stream will finish processing shortly after.
Describe the solution you'd like
Make skuber to return a WatchEvent of a new EventType. e.g.: WatchEvent[TooOldResourceError, Configmap]. So, we can discard those and not having to restart the stream because of the exception.
Describe alternatives you've considered
Instead of watchContinously, we can do something like:
Source.tick(0.seconds, 3.minutes, ()).flatMapConcat { _ =>
Source.lazyFuture(() => k8s.get[ConfigMap](cm.name))
}
to get the configmap in a different fashion, but we need to do some changes to know when there is new data in the configmap
Additional context
Conversation in discord:
https://discord.com/channels/862688166313852928/862688166313852931/1344622638135382026
Activity