Skip to content

Commit 64b2538

Browse files
stttsbertinatto
authored andcommitted
UPSTREAM: <carry>: create termination events
UPSTREAM: <carry>: apiserver: log new connections during termination UPSTREAM: <carry>: apiserver: create LateConnections events on events in the last 20% of graceful termination time UPSTREAM: <carry>: apiserver: log source in LateConnections event UPSTREAM: <carry>: apiserver: skip local IPs and probes for LateConnections UPSTREAM: <carry>: only create valid LateConnections/GracefulTermination events UPSTREAM: <carry>: kube-apiserver: log non-probe requests before ready UPSTREAM: <carry>: apiserver: create hasBeenReadyCh channel UPSTREAM: <carry>: kube-apiserver: log non-probe requests before ready UPSTREAM: <carry>: kube-apiserver: log non-probe requests before ready UPSTREAM: <carry>: fix termination event(s) validation failures UPSTREAM: <carry>: during the rebase collapse to create termination event it makes recording termination events a non-blocking operation. previously closing delayedStopCh might have been delayed on preserving data in the storage. the delayedStopCh is important as it signals the HTTP server to start the shutdown procedure. it also sets a hard timeout of 3 seconds for the storage layer since we are bypassing the API layer. UPSTREAM: <carry>: rename termination events to use lifecycleSignals OpenShift-Rebase-Source: 15b2d2e UPSTREAM: <carry>: extend termination events - we tie the shutdown events with the UID of the first (shutdown initiated), this provides us with a more deterministic way to compute shutdown duration from these events - move code snippets from the upstream file to openshift specific patch file, it reduces chance of code conflict
1 parent 7f50be3 commit 64b2538

File tree

5 files changed

+431
-2
lines changed

5 files changed

+431
-2
lines changed

pkg/controlplane/apiserver/config.go

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"k8s.io/kubernetes/openshift-kube-apiserver/admission/admissionenablement"
2929
"k8s.io/kubernetes/openshift-kube-apiserver/enablement"
3030
"k8s.io/kubernetes/openshift-kube-apiserver/openshiftkubeapiserver"
31+
eventstorage "k8s.io/kubernetes/pkg/registry/core/event/storage"
3132

3233
"k8s.io/apimachinery/pkg/api/meta"
3334
"k8s.io/apimachinery/pkg/runtime"
@@ -292,6 +293,13 @@ func CreateConfig(
292293
opts.Metrics.Apply()
293294
serviceaccount.RegisterMetrics()
294295

296+
var eventStorage *eventstorage.REST
297+
eventStorage, err := eventstorage.NewREST(genericConfig.RESTOptionsGetter, uint64(opts.EventTTL.Seconds()))
298+
if err != nil {
299+
return nil, nil, err
300+
}
301+
genericConfig.EventSink = eventRegistrySink{eventStorage}
302+
295303
config := &Config{
296304
Generic: genericConfig,
297305
Extra: Extra{
Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
Copyright 2024 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package apiserver
18+
19+
import (
20+
"context"
21+
"fmt"
22+
"time"
23+
24+
corev1 "k8s.io/api/core/v1"
25+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
26+
"k8s.io/apiserver/pkg/endpoints/request"
27+
genericapiserver "k8s.io/apiserver/pkg/server"
28+
"k8s.io/kubernetes/pkg/apis/core"
29+
v1 "k8s.io/kubernetes/pkg/apis/core/v1"
30+
eventstorage "k8s.io/kubernetes/pkg/registry/core/event/storage"
31+
)
32+
33+
// eventRegistrySink wraps an event registry in order to be used as direct event sync, without going through the API.
34+
type eventRegistrySink struct {
35+
*eventstorage.REST
36+
}
37+
38+
var _ genericapiserver.EventSink = eventRegistrySink{}
39+
40+
func (s eventRegistrySink) Create(v1event *corev1.Event) (*corev1.Event, error) {
41+
ctx := request.WithNamespace(request.WithRequestInfo(request.NewContext(), &request.RequestInfo{APIVersion: "v1"}), v1event.Namespace)
42+
// since we are bypassing the API set a hard timeout for the storage layer
43+
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
44+
defer cancel()
45+
46+
var event core.Event
47+
if err := v1.Convert_v1_Event_To_core_Event(v1event, &event, nil); err != nil {
48+
return nil, err
49+
}
50+
51+
obj, err := s.REST.Create(ctx, &event, nil, &metav1.CreateOptions{})
52+
if err != nil {
53+
return nil, err
54+
}
55+
ret, ok := obj.(*core.Event)
56+
if !ok {
57+
return nil, fmt.Errorf("expected corev1.Event, got %T", obj)
58+
}
59+
60+
var v1ret corev1.Event
61+
if err := v1.Convert_core_Event_To_v1_Event(ret, &v1ret, nil); err != nil {
62+
return nil, err
63+
}
64+
65+
return &v1ret, nil
66+
}

staging/src/k8s.io/apiserver/pkg/server/config.go

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,8 @@ import (
7272
utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
7373
flowcontrolrequest "k8s.io/apiserver/pkg/util/flowcontrol/request"
7474
"k8s.io/client-go/informers"
75+
"k8s.io/client-go/kubernetes"
76+
v1 "k8s.io/client-go/kubernetes/typed/core/v1"
7577
restclient "k8s.io/client-go/rest"
7678
basecompatibility "k8s.io/component-base/compatibility"
7779
"k8s.io/component-base/featuregate"
@@ -288,6 +290,9 @@ type Config struct {
288290
// rejected with a 429 status code and a 'Retry-After' response.
289291
ShutdownSendRetryAfter bool
290292

293+
// EventSink receives events about the life cycle of the API server, e.g. readiness, serving, signals and termination.
294+
EventSink EventSink
295+
291296
//===========================================================================
292297
// values below here are targets for removal
293298
//===========================================================================
@@ -724,6 +729,10 @@ func (c *Config) Complete(informers informers.SharedInformerFactory) CompletedCo
724729
c.DiscoveryAddresses = discovery.DefaultAddresses{DefaultAddress: c.ExternalAddress}
725730
}
726731

732+
if c.EventSink == nil {
733+
c.EventSink = nullEventSink{}
734+
}
735+
727736
AuthorizeClientBearerToken(c.LoopbackClientConfig, &c.Authentication, &c.Authorization)
728737

729738
if c.RequestInfoResolver == nil {
@@ -751,6 +760,22 @@ func (c *Config) Complete(informers informers.SharedInformerFactory) CompletedCo
751760
// Complete fills in any fields not set that are required to have valid data and can be derived
752761
// from other fields. If you're going to `ApplyOptions`, do that first. It's mutating the receiver.
753762
func (c *RecommendedConfig) Complete() CompletedConfig {
763+
if c.ClientConfig != nil {
764+
ref, err := eventReference()
765+
if err != nil {
766+
klog.Warningf("Failed to derive event reference, won't create events: %v", err)
767+
c.EventSink = nullEventSink{}
768+
} else {
769+
ns := ref.Namespace
770+
if len(ns) == 0 {
771+
ns = "default"
772+
}
773+
c.EventSink = clientEventSink{
774+
&v1.EventSinkImpl{Interface: kubernetes.NewForConfigOrDie(c.ClientConfig).CoreV1().Events(ns)},
775+
}
776+
}
777+
}
778+
754779
return c.Config.Complete(c.SharedInformerFactory)
755780
}
756781

@@ -855,7 +880,19 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
855880
FeatureGate: c.FeatureGate,
856881

857882
muxAndDiscoveryCompleteSignals: map[string]<-chan struct{}{},
883+
884+
OpenShiftGenericAPIServerPatch: OpenShiftGenericAPIServerPatch{
885+
eventSink: c.EventSink,
886+
},
887+
}
888+
889+
ref, err := eventReference()
890+
if err != nil {
891+
klog.Warningf("Failed to derive event reference, won't create events: %v", err)
892+
s.OpenShiftGenericAPIServerPatch.eventSink = nullEventSink{}
858893
}
894+
s.RegisterDestroyFunc(c.EventSink.Destroy)
895+
s.eventRef = ref
859896

860897
manager := c.AggregatedDiscoveryGroupManager
861898
if manager == nil {
@@ -1063,6 +1100,8 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
10631100
handler = genericapifilters.WithRequestDeadline(handler, c.AuditBackend, c.AuditPolicyRuleEvaluator,
10641101
c.LongRunningFunc, c.Serializer, c.RequestTimeout)
10651102
handler = genericfilters.WithWaitGroup(handler, c.LongRunningFunc, c.NonLongRunningRequestWaitGroup)
1103+
handler = WithNonReadyRequestLogging(handler, c.lifecycleSignals.HasBeenReady)
1104+
handler = WithLateConnectionFilter(handler)
10661105
if c.ShutdownWatchTerminationGracePeriod > 0 {
10671106
handler = genericfilters.WithWatchTerminationDuringShutdown(handler, c.lifecycleSignals, c.WatchRequestWaitGroup)
10681107
}

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030

3131
"golang.org/x/time/rate"
3232
apidiscoveryv2 "k8s.io/api/apidiscovery/v2"
33+
corev1 "k8s.io/api/core/v1"
3334
"k8s.io/apimachinery/pkg/api/meta"
3435
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3536
"k8s.io/apimachinery/pkg/runtime"
@@ -296,6 +297,9 @@ type GenericAPIServer struct {
296297
// This grace period is orthogonal to other grace periods, and
297298
// it is not overridden by any other grace period.
298299
ShutdownWatchTerminationGracePeriod time.Duration
300+
301+
// OpenShift patch
302+
OpenShiftGenericAPIServerPatch
299303
}
300304

301305
// DelegationTarget is an interface which allows for composition of API servers with top level handling that works
@@ -548,7 +552,10 @@ func (s preparedGenericAPIServer) RunWithContext(ctx context.Context) error {
548552

549553
go func() {
550554
defer delayedStopCh.Signal()
551-
defer klog.V(1).InfoS("[graceful-termination] shutdown event", "name", delayedStopCh.Name())
555+
defer func() {
556+
klog.V(1).InfoS("[graceful-termination] shutdown event", "name", delayedStopCh.Name())
557+
s.Eventf(corev1.EventTypeNormal, delayedStopCh.Name(), "The minimal shutdown duration of %v finished", s.ShutdownDelayDuration)
558+
}()
552559

553560
<-stopCh
554561

@@ -557,10 +564,28 @@ func (s preparedGenericAPIServer) RunWithContext(ctx context.Context) error {
557564
// and stop sending traffic to this server.
558565
shutdownInitiatedCh.Signal()
559566
klog.V(1).InfoS("[graceful-termination] shutdown event", "name", shutdownInitiatedCh.Name())
567+
s.Eventf(corev1.EventTypeNormal, shutdownInitiatedCh.Name(), "Received signal to terminate, becoming unready, but keeping serving")
560568

561569
time.Sleep(s.ShutdownDelayDuration)
562570
}()
563571

572+
lateStopCh := make(chan struct{})
573+
if s.ShutdownDelayDuration > 0 {
574+
go func() {
575+
defer close(lateStopCh)
576+
577+
<-stopCh
578+
579+
time.Sleep(s.ShutdownDelayDuration * 8 / 10)
580+
}()
581+
}
582+
583+
s.SecureServingInfo.Listener = &terminationLoggingListener{
584+
Listener: s.SecureServingInfo.Listener,
585+
lateStopCh: lateStopCh,
586+
}
587+
unexpectedRequestsEventf.Store(s.Eventf)
588+
564589
// close socket after delayed stopCh
565590
shutdownTimeout := s.ShutdownTimeout
566591
if s.ShutdownSendRetryAfter {
@@ -609,13 +634,17 @@ func (s preparedGenericAPIServer) RunWithContext(ctx context.Context) error {
609634
<-listenerStoppedCh
610635
httpServerStoppedListeningCh.Signal()
611636
klog.V(1).InfoS("[graceful-termination] shutdown event", "name", httpServerStoppedListeningCh.Name())
637+
s.Eventf(corev1.EventTypeNormal, httpServerStoppedListeningCh.Name(), "HTTP Server has stopped listening")
612638
}()
613639

614640
// we don't accept new request as soon as both ShutdownDelayDuration has
615641
// elapsed and preshutdown hooks have completed.
616642
preShutdownHooksHasStoppedCh := s.lifecycleSignals.PreShutdownHooksStopped
617643
go func() {
618-
defer klog.V(1).InfoS("[graceful-termination] shutdown event", "name", notAcceptingNewRequestCh.Name())
644+
defer func() {
645+
klog.V(1).InfoS("[graceful-termination] shutdown event", "name", notAcceptingNewRequestCh.Name())
646+
s.Eventf(corev1.EventTypeNormal, drainedCh.Name(), "All non long-running request(s) in-flight have drained")
647+
}()
619648
defer notAcceptingNewRequestCh.Signal()
620649

621650
// wait for the delayed stopCh before closing the handler chain
@@ -702,6 +731,7 @@ func (s preparedGenericAPIServer) RunWithContext(ctx context.Context) error {
702731
defer func() {
703732
preShutdownHooksHasStoppedCh.Signal()
704733
klog.V(1).InfoS("[graceful-termination] pre-shutdown hooks completed", "name", preShutdownHooksHasStoppedCh.Name())
734+
s.Eventf(corev1.EventTypeNormal, "TerminationPreShutdownHooksFinished", "All pre-shutdown hooks have been finished")
705735
}()
706736
err = s.RunPreShutdownHooks()
707737
}()
@@ -722,6 +752,8 @@ func (s preparedGenericAPIServer) RunWithContext(ctx context.Context) error {
722752
<-stoppedCh
723753

724754
klog.V(1).Info("[graceful-termination] apiserver is exiting")
755+
s.Eventf(corev1.EventTypeNormal, "TerminationGracefulTerminationFinished", "All pending requests processed")
756+
725757
return nil
726758
}
727759

0 commit comments

Comments
 (0)