ronantakizawa/codereview-bench huggingface dataset:
Repository: kubernetes/kubernetes
File: test/utils/ktesting/stepcontext_test.go (Go)

Code before:
`,
		},
		"progress": {
			cb: func(tCtx TContext) {
				tCtx = WithStep(tCtx, "step")
				var buffer bytes.Buffer
				oldOut := defaultProgressReporter.setOutput(&buffer)
				defer defaultProgressReporter.setOutput(oldOut)
				remove := tCtx.Value("GINKGO_SPEC_CONTEXT").(ginkgoReporter).AttachProgressReporter(func() string { return "hello world" })
				defer remove()
				defaultSignalChannel <- os.Interrupt
				// No good way to sync here, so let's just wait.
				time.Sleep(5 * time.Second)
				defaultProgressReporter.setOutput(oldOut)
				tCtx.Log(buffer.String())

				noSuchValue := tCtx.Value("some other key")
				assert.Nil(tCtx, noSuchValue, "value for unknown context value key")
			},
			expectTrace: `(LOG) <klog header>: step: You requested a progress report.
	
	step: hello world
`,
			expectDuration: 5 * time.Second,
		},
	} {
		t.Run(name, func(t *testing.T) {
			tc.run(t)
		})
	}
}


Review comment (suggestion):
```suggestion
	removeReporter := tCtx.Value(ginkgoSpecContextKey).(ginkgoReporter).AttachProgressReporter(func() string { return "hello world" })
```

Code after:
`,
		},
	} {
		tc := tc
		t.Run(name, func(t *testing.T) {
			tc.run(t)
		})
	}
}

func TestProgressReport(t *testing.T) {
	t.Cleanup(func() {
		goleak.VerifyNone(t)
	})

	oldOut := defaultProgressReporter.out
	reportStream := newOutputStream()
	defaultProgressReporter.out = reportStream
	t.Cleanup(func() {
		defaultProgressReporter.out = oldOut
	})

	// This must use a real testing.T, otherwise Init doesn't initialize signal handling.
	tCtx := Init(t)
	tCtx = WithStep(tCtx, "step")
	removeReporter := tCtx.Value("GINKGO_SPEC_CONTEXT").(ginkgoReporter).AttachProgressReporter(func() string { return "hello world" })
	defer removeReporter()
	tCtx.Expect(tCtx.Value("some other key")).To(gomega.BeNil(), "value for unknown context value key")

	// Trigger report and wait for it.
	defaultProgressReporter.progressChannel <- os.Interrupt
	report := <-reportStream.stream
	tCtx.Expect(report).To(gomega.Equal(`You requested a progress report.

step: hello world
`), "report")
}

// outputStream forwards exactly one Write call to a stream.
// A second Write call is an error and will panic.
type outputStream struct {
	stream chan string
}

var _ io.Writer = &outputStream{}

func newOutputStream() *outputStream {
	return &outputStream{
		stream: make(chan string),
	}
}
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apimachinery/pkg/util/validation/field/errors.go (Go)

Code before:
// ExtractCoveredByDeclarative returns a new ErrorList containing only the errors that should be covered by declarative validation.
func (list ErrorList) ExtractCoveredByDeclarative() ErrorList {
	newList := ErrorList{}
	for _, err := range list {
		if err.CoveredByDeclarative {
			newList = append(newList, err)
		}
	}
	return newList
}

// MarkDeclarativeNative marks the error as originating from a declarative-native validation.
func (e *Error) MarkDeclarativeNative() *Error {
	e.DeclarativeNative = true
	return e
}

// MarkDeclarativeNative marks all errors in the list as originating from declarative-native validations.
func (list ErrorList) MarkDeclarativeNative() ErrorList {
	for _, err := range list {
		err.DeclarativeNative = true
	}
	return list
}

func (e *Error) MarkOriginatingFromImperative() *Error {
	e.OriginatingFromImperative = true
	return e
}

func (list ErrorList) MarkOriginatingFromImperative() ErrorList {
	for _, err := range list {
		err.OriginatingFromImperative = true
	}
	return list
}

// RemoveCoveredByDeclarative returns a new ErrorList containing only the errors that should not be covered by declarative validation.
func (list ErrorList) RemoveCoveredByDeclarative() ErrorList {
	newList := ErrorList{}
	for _, err := range list {
		if !err.CoveredByDeclarative {
			newList = append(newList, err)
		}
	}
	return newList
}


Review comment (suggestion):
Can we add a test for this new method in: `staging/src/k8s.io/apimachinery/pkg/util/validation/field/errors_test.go`

Code after:
	for _, err := range list {
		msg := fmt.Sprintf("%v", err)
		if errorMsgs.Has(msg) {
			continue
		}
		errorMsgs.Insert(msg)
		errs = append(errs, err)
	}
	return utilerrors.NewAggregate(errs)
}

func fromAggregate(agg utilerrors.Aggregate) ErrorList {
	errs := agg.Errors()
	list := make(ErrorList, len(errs))
	for i := range errs {
		list[i] = errs[i].(*Error)
	}
	return list
}

// Filter removes items from the ErrorList that match the provided fns.
func (list ErrorList) Filter(fns ...utilerrors.Matcher) ErrorList {
	err := utilerrors.FilterOut(list.ToAggregate(), fns...)
	if err == nil {
		return nil
	}
	// FilterOut takes an Aggregate and returns an Aggregate
	return fromAggregate(err.(utilerrors.Aggregate))
}

// ExtractCoveredByDeclarative returns a new ErrorList containing only the errors that should be covered by declarative validation.
func (list ErrorList) ExtractCoveredByDeclarative() ErrorList {
	newList := ErrorList{}
	for _, err := range list {
		if err.CoveredByDeclarative {
			newList = append(newList, err)
		}
	}
	return newList
}

// MarkAlpha marks the error as an alpha validation error.
func (e *Error) MarkAlpha() *Error {
	e.ValidationStabilityLevel = stabilityLevelAlpha
	return e
}

// MarkAlpha marks the errors as alpha validation errors.
func (list ErrorList) MarkAlpha() ErrorList {
	for _, err := range list {
		err.ValidationStabilityLevel = stabilityLevelAlpha
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apimachinery/pkg/util/validation/field/errors.go (Go)

Code before:
			newList = append(newList, err)
		}
	}
	return newList
}

// MarkDeclarativeNative marks the error as originating from a declarative-native validation.
func (e *Error) MarkDeclarativeNative() *Error {
	e.DeclarativeNative = true
	return e
}

// MarkDeclarativeNative marks all errors in the list as originating from declarative-native validations.
func (list ErrorList) MarkDeclarativeNative() ErrorList {
	for _, err := range list {
		err.DeclarativeNative = true
	}
	return list
}

func (e *Error) MarkOriginatingFromImperative() *Error {
	e.OriginatingFromImperative = true
	return e
}

func (list ErrorList) MarkOriginatingFromImperative() ErrorList {
	for _, err := range list {
		err.OriginatingFromImperative = true
	}
	return list
}

// RemoveCoveredByDeclarative returns a new ErrorList containing only the errors that should not be covered by declarative validation.
func (list ErrorList) RemoveCoveredByDeclarative() ErrorList {
	newList := ErrorList{}
	for _, err := range list {
		if !err.CoveredByDeclarative {
			newList = append(newList, err)
		}
	}
	return newList
}


Review comment (suggestion):
Can we add a test for this new method in: `staging/src/k8s.io/apimachinery/pkg/util/validation/field/errors_test.go`

Code after:
		errorMsgs.Insert(msg)
		errs = append(errs, err)
	}
	return utilerrors.NewAggregate(errs)
}

func fromAggregate(agg utilerrors.Aggregate) ErrorList {
	errs := agg.Errors()
	list := make(ErrorList, len(errs))
	for i := range errs {
		list[i] = errs[i].(*Error)
	}
	return list
}

// Filter removes items from the ErrorList that match the provided fns.
func (list ErrorList) Filter(fns ...utilerrors.Matcher) ErrorList {
	err := utilerrors.FilterOut(list.ToAggregate(), fns...)
	if err == nil {
		return nil
	}
	// FilterOut takes an Aggregate and returns an Aggregate
	return fromAggregate(err.(utilerrors.Aggregate))
}

// ExtractCoveredByDeclarative returns a new ErrorList containing only the errors that should be covered by declarative validation.
func (list ErrorList) ExtractCoveredByDeclarative() ErrorList {
	newList := ErrorList{}
	for _, err := range list {
		if err.CoveredByDeclarative {
			newList = append(newList, err)
		}
	}
	return newList
}

// MarkAlpha marks the error as an alpha validation error.
func (e *Error) MarkAlpha() *Error {
	e.ValidationStabilityLevel = stabilityLevelAlpha
	return e
}

// MarkAlpha marks the errors as alpha validation errors.
func (list ErrorList) MarkAlpha() ErrorList {
	for _, err := range list {
		err.ValidationStabilityLevel = stabilityLevelAlpha
	}
	return list
}

// MarkBeta marks the error as a beta validation error.
Repository: kubernetes/kubernetes
File: pkg/kubelet/cm/dra/manager_test.go (Go)

Code before:
}

// genTestClaimInfo generates claim info object
func genTestClaimInfo(claimUID types.UID, podUIDs []string, prepared bool) *ClaimInfo {
	return &ClaimInfo{
		ClaimInfoState: state.ClaimInfoState{
			ClaimUID:  claimUID,
			ClaimName: claimName,
			Namespace: namespace,
			PodUIDs:   sets.New[string](podUIDs...),
			DriverState: map[string]state.DriverState{
				driverName: {
					Devices: []state.Device{{
						PoolName:     poolName,
						DeviceName:   deviceName,
						RequestNames: []string{requestName},
						CDIDeviceIDs: []string{cdiID},
					}},
				},
			},
		},
		prepared: prepared,
	}
}

// genTestClaimInfoWithShareID generates claim info object with ShareID set on the device.
// This is used to test health status reporting for shared/pooled devices.
func genTestClaimInfoWithShareID(claimUID types.UID, podUIDs []string, prepared bool, shareID *types.UID) *ClaimInfo {
	return &ClaimInfo{
		ClaimInfoState: state.ClaimInfoState{
			ClaimUID:  claimUID,
			ClaimName: claimName,
			Namespace: namespace,
			PodUIDs:   sets.New[string](podUIDs...),
			DriverState: map[string]state.DriverState{
				driverName: {
					Devices: []state.Device{{
						PoolName:     poolName,
						DeviceName:   deviceName,
						ShareID:      shareID,
						RequestNames: []string{requestName},
						CDIDeviceIDs: []string{cdiID},
					}},
				},
			},
		},
		prepared: prepared,
	}
}

// genTestClaimInfoWithExtendedResource generates claim info object

Review comment (suggestion):
I think we can just extend `genTestClaimInfo` with `shareID`, and pass in nil in other cases it's called

Code after:
}

// genTestClaimInfo generates claim info object
func genTestClaimInfo(claimUID types.UID, podUIDs []string, prepared bool, shareID *types.UID) *ClaimInfo {
	return &ClaimInfo{
		ClaimInfoState: state.ClaimInfoState{
			ClaimUID:  claimUID,
			ClaimName: claimName,
			Namespace: namespace,
			PodUIDs:   sets.New[string](podUIDs...),
			DriverState: map[string]state.DriverState{
				driverName: {
					Devices: []state.Device{{
						PoolName:     poolName,
						DeviceName:   deviceName,
						ShareID:      shareID,
						RequestNames: []string{requestName},
						CDIDeviceIDs: []string{cdiID},
					}},
				},
			},
		},
		prepared: prepared,
	}
}

// genTestClaimInfoWithExtendedResource generates claim info object
func genTestClaimInfoWithExtendedResource(podUIDs []string, prepared bool) *ClaimInfo {
	return &ClaimInfo{
		ClaimInfoState: state.ClaimInfoState{
			ClaimUID:  claimUID,
			ClaimName: claimName,
			Namespace: namespace,
			PodUIDs:   sets.New[string](podUIDs...),
			DriverState: map[string]state.DriverState{
				driverName: {
					Devices: []state.Device{{
						PoolName:     poolName,
						DeviceName:   deviceName,
						RequestNames: []string{"container-0-request-0"},
						CDIDeviceIDs: []string{cdiID},
					}},
				},
			},
		},
		prepared: prepared,
	}
}

// genClaimInfoState generates claim info state object
func genClaimInfoState(cdiDeviceID string) state.ClaimInfoState {
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apiextensions-apiserver/test/integration/conversion/conversion_test.go (Go)

Code before:
func jsonPtr(x interface{}) *apiextensionsv1.JSON {
	bs, err := json.Marshal(x)
	if err != nil {
		panic(err)
	}
	ret := apiextensionsv1.JSON{Raw: bs}
	return &ret
}

func servedVersions(versions []apiextensionsv1.CustomResourceDefinitionVersion) (served []apiextensionsv1.CustomResourceDefinitionVersion) {
	for _, v := range versions {
		if v.Served {
			served = append(served, v)
		}
	}

	return served
}

func TestWebhookConversion_WhitespaceCABundleEtcdBypass(t *testing.T) {
	// Setup server and clients
	apiServerTearDown, config, options, err := fixtures.StartDefaultServer(t)
	if err != nil {
		t.Fatal(err)
	}
	// defer apiServerTearDown()

	apiExtensionsClient, err := clientset.NewForConfig(config)
	if err != nil {
		t.Fatal(err)
	}
	dynamicClient, err := dynamic.NewForConfig(config)
	if err != nil {
		t.Fatal(err)
	}

	crd := multiVersionFixture.DeepCopy()

	RESTOptionsGetter := serveroptions.NewCRDRESTOptionsGetter(*options.RecommendedOptions.Etcd, nil, nil)
	restOptions, err := RESTOptionsGetter.GetRESTOptions(schema.GroupResource{Group: crd.Spec.Group, Resource: crd.Spec.Names.Plural}, nil)
	if err != nil {
		t.Fatal(err)
	}
	etcdClient, _, err := storage.GetEtcdClients(restOptions.StorageConfig.Transport)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		err = etcdClient.Close()
		if err != nil {
			t.Fatal(err)

Review comment (refactor):
Let's remove this commented out line entirely.  I confused me for a moment until I realized that  the defer has been replaced by `t.Cleanup`

Code after:
func jsonPtr(x interface{}) *apiextensionsv1.JSON {
	bs, err := json.Marshal(x)
	if err != nil {
		panic(err)
	}
	ret := apiextensionsv1.JSON{Raw: bs}
	return &ret
}

func servedVersions(versions []apiextensionsv1.CustomResourceDefinitionVersion) (served []apiextensionsv1.CustomResourceDefinitionVersion) {
	for _, v := range versions {
		if v.Served {
			served = append(served, v)
		}
	}

	return served
}

func TestWebhookConversion_WhitespaceCABundleEtcdBypass(t *testing.T) {
	// Setup server and clients
	apiServerTearDown, config, options, err := fixtures.StartDefaultServer(t)
	if err != nil {
		t.Fatal(err)
	}

	apiExtensionsClient, err := clientset.NewForConfig(config)
	if err != nil {
		t.Fatal(err)
	}
	dynamicClient, err := dynamic.NewForConfig(config)
	if err != nil {
		t.Fatal(err)
	}

	crd := multiVersionFixture.DeepCopy()

	RESTOptionsGetter := serveroptions.NewCRDRESTOptionsGetter(*options.RecommendedOptions.Etcd, nil, nil)
	restOptions, err := RESTOptionsGetter.GetRESTOptions(schema.GroupResource{Group: crd.Spec.Group, Resource: crd.Spec.Names.Plural}, nil)
	if err != nil {
		t.Fatal(err)
	}
	etcdClient, _, err := storage.GetEtcdClients(restOptions.StorageConfig.Transport)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		err = etcdClient.Close()
		if err != nil {
			t.Fatal(err)
		}
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/kubectl/pkg/cmd/rollout/rollout_undo.go (Go)

Code before:

// RunUndo performs the execution of 'rollout undo' sub command
func (o *UndoOptions) RunUndo() error {
	r := o.Builder().
		WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
		NamespaceParam(o.Namespace).DefaultNamespace().
		LabelSelectorParam(o.LabelSelector).
		FilenameParam(o.EnforceNamespace, &o.FilenameOptions).
		ResourceTypeOrNameArgs(true, o.Resources...).
		ContinueOnError().
		Latest().
		Flatten().
		Do()
	if err := r.Err(); err != nil {
		return err
	}

	err := r.Visit(func(info *resource.Info, err error) error {
		if err != nil {
			return err
		}

		metadata, _ := meta.Accessor(info.Object)
		annotationMap := metadata.GetAnnotations()
		if _, ok := annotationMap[corev1.LastAppliedConfigAnnotation]; ok {
			fmt.Fprintf(o.ErrOut, warningRollbackApplyManagedResource, info.ObjectName(), corev1.LastAppliedConfigAnnotation)
		}

		rollbacker, err := polymorphichelpers.RollbackerFn(o.RESTClientGetter, info.ResourceMapping())
		if err != nil {
			return err
		}

		result, err := rollbacker.Rollback(info.Object, nil, o.ToRevision, o.DryRunStrategy)
		if err != nil {
			return err
		}

		printer, err := o.ToPrinter(result)
		if err != nil {
			return err
		}

		return printer.PrintObj(info.Object, o.Out)
	})

	return err
}


Review comment (suggestion):
I'm certain the linter will complain, so let's do:
```suggestion
			fmt.Fprintf(o.ErrOut, warningRollbackApplyManagedResource, info.ObjectName(), corev1.LastAppliedConfigAnnotation) //nolint:errcheck
```

Code after:
	return nil
}

// RunUndo performs the execution of 'rollout undo' sub command
func (o *UndoOptions) RunUndo() error {
	r := o.Builder().
		WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
		NamespaceParam(o.Namespace).DefaultNamespace().
		LabelSelectorParam(o.LabelSelector).
		FilenameParam(o.EnforceNamespace, &o.FilenameOptions).
		ResourceTypeOrNameArgs(true, o.Resources...).
		ContinueOnError().
		Latest().
		Flatten().
		Do()
	if err := r.Err(); err != nil {
		return err
	}

	err := r.Visit(func(info *resource.Info, err error) error {
		if err != nil {
			return err
		}

		metadata, err := meta.Accessor(info.Object)
		if err != nil {
			return err
		}

		annotationMap := metadata.GetAnnotations()
		if _, ok := annotationMap[corev1.LastAppliedConfigAnnotation]; ok {
			fmt.Fprintf(o.ErrOut, warningRollbackApplyManagedResource, info.ObjectName(), corev1.LastAppliedConfigAnnotation, o.CmdBaseName) //nolint:errcheck
		}

		rollbacker, err := polymorphichelpers.RollbackerFn(o.RESTClientGetter, info.ResourceMapping())
		if err != nil {
			return err
		}

		result, err := rollbacker.Rollback(info.Object, nil, o.ToRevision, o.DryRunStrategy)
		if err != nil {
			return err
		}

		printer, err := o.ToPrinter(result)
		if err != nil {
			return err
		}

		return printer.PrintObj(info.Object, o.Out)
	})
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/kubectl/pkg/cmd/rollout/rollout_undo.go (Go)

Code before:
	}
	return nil
}

// RunUndo performs the execution of 'rollout undo' sub command
func (o *UndoOptions) RunUndo() error {
	r := o.Builder().
		WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
		NamespaceParam(o.Namespace).DefaultNamespace().
		LabelSelectorParam(o.LabelSelector).
		FilenameParam(o.EnforceNamespace, &o.FilenameOptions).
		ResourceTypeOrNameArgs(true, o.Resources...).
		ContinueOnError().
		Latest().
		Flatten().
		Do()
	if err := r.Err(); err != nil {
		return err
	}

	err := r.Visit(func(info *resource.Info, err error) error {
		if err != nil {
			return err
		}

		metadata, _ := meta.Accessor(info.Object)
		annotationMap := metadata.GetAnnotations()
		if _, ok := annotationMap[corev1.LastAppliedConfigAnnotation]; ok {
			fmt.Fprintf(o.ErrOut, warningRollbackApplyManagedResource, info.ObjectName(), corev1.LastAppliedConfigAnnotation, o.CmdBaseName) //nolint:errcheck
		}

		rollbacker, err := polymorphichelpers.RollbackerFn(o.RESTClientGetter, info.ResourceMapping())
		if err != nil {
			return err
		}

		result, err := rollbacker.Rollback(info.Object, nil, o.ToRevision, o.DryRunStrategy)
		if err != nil {
			return err
		}

		printer, err := o.ToPrinter(result)
		if err != nil {
			return err
		}

		return printer.PrintObj(info.Object, o.Out)
	})

	return err
}

Review comment (suggestion):
Sorry I missed this last time, you need to handle this error from `meta.Accessor`, if this happens, we'll just don't print the warning. 

Code after:
	}
	return nil
}

// RunUndo performs the execution of 'rollout undo' sub command
func (o *UndoOptions) RunUndo() error {
	r := o.Builder().
		WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
		NamespaceParam(o.Namespace).DefaultNamespace().
		LabelSelectorParam(o.LabelSelector).
		FilenameParam(o.EnforceNamespace, &o.FilenameOptions).
		ResourceTypeOrNameArgs(true, o.Resources...).
		ContinueOnError().
		Latest().
		Flatten().
		Do()
	if err := r.Err(); err != nil {
		return err
	}

	err := r.Visit(func(info *resource.Info, err error) error {
		if err != nil {
			return err
		}

		metadata, err := meta.Accessor(info.Object)
		if err != nil {
			return err
		}

		annotationMap := metadata.GetAnnotations()
		if _, ok := annotationMap[corev1.LastAppliedConfigAnnotation]; ok {
			fmt.Fprintf(o.ErrOut, warningRollbackApplyManagedResource, info.ObjectName(), corev1.LastAppliedConfigAnnotation, o.CmdBaseName) //nolint:errcheck
		}

		rollbacker, err := polymorphichelpers.RollbackerFn(o.RESTClientGetter, info.ResourceMapping())
		if err != nil {
			return err
		}

		result, err := rollbacker.Rollback(info.Object, nil, o.ToRevision, o.DryRunStrategy)
		if err != nil {
			return err
		}

		printer, err := o.ToPrinter(result)
		if err != nil {
			return err
		}

		return printer.PrintObj(info.Object, o.Out)
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go (Go)

Code before:
					httplog.AddKeyValue(req.Context(), "watchlist_init_latency", initLatency)
				}
			}
		}
	}
}

// HandleWS serves a series of encoded events over a websocket connection.
func (s *WatchServer) HandleWS(ws *websocket.Conn) {
	ctx := ws.Request().Context()
	logger := klog.FromContext(ctx)

	defer func() {
		if s.MemoryAllocator != nil {
			runtime.AllocatorPool.Put(s.MemoryAllocator)
		}
	}()

	defer ws.Close()
	done := make(chan struct{})
	// ensure the connection times out
	timeoutCh, cleanup := s.TimeoutFactory.TimeoutCh()
	defer cleanup()

	go func() {
		defer utilruntime.HandleCrashWithContext(ctx)
		// This blocks until the connection is closed.
		// Client should not send anything.
		wsstream.IgnoreReceivesWithLogger(logger, ws, 0)
		// Once the client closes, we should also close
		close(done)
	}()

	framer := newWebsocketFramer(ws, s.UseTextFraming)

	gvr := s.Scope.Resource
	watchEncoder := newWatchEncoder(ctx, gvr, s.EmbeddedEncoder, s.Encoder, framer)
	ch := s.Watching.ResultChan()

	for {
		select {
		case <-done:
			return
		case <-timeoutCh:
			return
		case event, ok := <-ch:
			if !ok {
				// End of results.
				return
			}


Review comment (bug):
```suggestion
		defer utilruntime.HandleCrashWithLogger(logger)
```

Then the goroutine function only needs to bind the logger.


Code after:
					klog.V(3).InfoS("WatchList initial events sent", "path", req.URL.Path, "auditID", auditID, "initLatency", initLatency)
					httplog.AddKeyValue(req.Context(), "watchlist_init_latency", initLatency)
				}
			}
		}
	}
}

// HandleWS serves a series of encoded events over a websocket connection.
func (s *WatchServer) HandleWS(ws *websocket.Conn) {
	ctx := ws.Request().Context()
	logger := klog.FromContext(ctx)

	defer func() {
		if s.MemoryAllocator != nil {
			runtime.AllocatorPool.Put(s.MemoryAllocator)
		}
	}()

	defer ws.Close()
	done := make(chan struct{})
	// ensure the connection times out
	timeoutCh, cleanup := s.TimeoutFactory.TimeoutCh()
	defer cleanup()

	go func() {
		defer utilruntime.HandleCrashWithLogger(logger)
		// This blocks until the connection is closed.
		// Client should not send anything.
		wsstream.IgnoreReceivesWithLogger(logger, ws, 0)
		// Once the client closes, we should also close
		close(done)
	}()

	framer := newWebsocketFramer(ws, s.UseTextFraming)

	gvr := s.Scope.Resource
	watchEncoder := newWatchEncoder(ctx, gvr, s.EmbeddedEncoder, s.Encoder, framer)
	ch := s.Watching.ResultChan()

	for {
		select {
		case <-done:
			return
		case <-timeoutCh:
			return
		case event, ok := <-ch:
			if !ok {
				// End of results.
				return
			}