github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cri: allow GetContainerEvents to pass a context

Browse Source 
The context can be used for timeout purposes for example, not really for
Kubernetes but other consumers like cri-tools.

Signed-off-by: Sascha Grunert <sgrunert@redhat.com>
This commit is contained in:
Sascha Grunert
2024-07-02 12:03:34 +02:00
parent d729af9446
commit ae8f6f002d
5 changed files with 8 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/kubelet/kuberuntime/instrumented_services.go
View File

@@ -335,11 +335,11 @@ func (in instrumentedRuntimeService) CheckpointContainer(ctx context.Context, op
return err return err
} }
func (in instrumentedRuntimeService) GetContainerEvents(containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error { func (in instrumentedRuntimeService) GetContainerEvents(ctx context.Context, containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error {
const operation = "get_container_events" const operation = "get_container_events"
defer recordOperation(operation, time.Now()) defer recordOperation(operation, time.Now())
err := in.service.GetContainerEvents(containerEventsCh, connectionEstablishedCallback) err := in.service.GetContainerEvents(ctx, containerEventsCh, connectionEstablishedCallback)
recordError(operation, err) recordError(operation, err)
return err return err
} }

3
pkg/kubelet/pleg/evented.go
View File

@@ -17,6 +17,7 @@ limitations under the License.
package pleg package pleg
import ( import (
"context"
"fmt" "fmt"
"sync" "sync"
"time" "time"
@@ -192,7 +193,7 @@ func (e *EventedPLEG) watchEventsChannel() {
} }
} }
err := e.runtimeService.GetContainerEvents(containerEventsResponseCh, func(runtimeapi.RuntimeService_GetContainerEventsClient) { err := e.runtimeService.GetContainerEvents(context.Background(), containerEventsResponseCh, func(runtimeapi.RuntimeService_GetContainerEventsClient) {
metrics.EventedPLEGConn.Inc() metrics.EventedPLEGConn.Inc()
}) })
if err != nil { if err != nil {

2
staging/src/k8s.io/cri-api/pkg/apis/services.go
View File

@@ -61,7 +61,7 @@ type ContainerManager interface {
// CheckpointContainer checkpoints a container // CheckpointContainer checkpoints a container
CheckpointContainer(ctx context.Context, options *runtimeapi.CheckpointContainerRequest) error CheckpointContainer(ctx context.Context, options *runtimeapi.CheckpointContainerRequest) error
// GetContainerEvents gets container events from the CRI runtime // GetContainerEvents gets container events from the CRI runtime
GetContainerEvents(containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error GetContainerEvents(ctx context.Context, containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error
} }
// PodSandboxManager contains methods for operating on PodSandboxes. The methods // PodSandboxManager contains methods for operating on PodSandboxes. The methods

2
staging/src/k8s.io/cri-api/pkg/apis/testing/fake_runtime_service.go
View File

@@ -716,7 +716,7 @@ func (r *FakeRuntimeService) CheckpointContainer(_ context.Context, options *run
return nil return nil
} }
func (f *FakeRuntimeService) GetContainerEvents(containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error { func (f *FakeRuntimeService) GetContainerEvents(ctx context.Context, containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error {
return nil return nil
} }

4
staging/src/k8s.io/cri-client/pkg/remote_runtime.go
View File

@@ -822,8 +822,8 @@ func (r *remoteRuntimeService) CheckpointContainer(ctx context.Context, options
return nil return nil
} }
func (r *remoteRuntimeService) GetContainerEvents(containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error { func (r *remoteRuntimeService) GetContainerEvents(ctx context.Context, containerEventsCh chan *runtimeapi.ContainerEventResponse, connectionEstablishedCallback func(runtimeapi.RuntimeService_GetContainerEventsClient)) error {
containerEventsStreamingClient, err := r.runtimeClient.GetContainerEvents(context.Background(), &runtimeapi.GetEventsRequest{}) containerEventsStreamingClient, err := r.runtimeClient.GetContainerEvents(ctx, &runtimeapi.GetEventsRequest{})
if err != nil { if err != nil {
r.logErr(err, "GetContainerEvents failed to get streaming client") r.logErr(err, "GetContainerEvents failed to get streaming client")
return err return err