Method startInotify

pkg/hostagent/inotify.go:29–85

(ctx context.Context)

)

func (a *HostAgent) startInotify(ctx context.Context) error {
	mountWatchCh := make(chan notify.EventInfo, 128)
	if err := a.setupWatchers(mountWatchCh); err != nil {
		return err
	}
	// notify.Watch allocates per-call kernel watchers and an internal reader
	// goroutine; without notify.Stop they leak for the lifetime of the process.
	defer notify.Stop(mountWatchCh)

	// PostInotify returns a usable stream wrapper before the server-side
	// handler is installed, so Sends issued during the gap fail with EOF.
	select {
	case <-a.guestAgentAliveCh:
	case <-ctx.Done():
		return nil
	}

	client, err := a.getOrCreateClient(ctx)
	if err != nil {
		return fmt.Errorf("inotify: failed to obtain guest agent client: %w", err)
	}
	inotifyClient, err := client.Inotify(ctx)
	if err != nil {
		return err
	}
	// Finalize the gRPC client-stream so the guest agent's PostInotify handler
	// can return instead of staying parked on a half-open stream.
	defer func() { _ = inotifyClient.CloseSend() }()

	for {
		select {
		case <-ctx.Done():
			return nil
		case watchEvent := <-mountWatchCh:
			watchPath := watchEvent.Path()
			stat, err := os.Stat(watchPath)
			if err != nil {
				continue
			}

			if filterEvents(watchEvent, stat) {
				continue
			}

			watchPath = translateToGuestPath(watchPath, mountSymlinks, mountLocations)

			utcTimestamp := timestamppb.New(stat.ModTime().UTC())
			event := &guestagentapi.Inotify{MountPath: watchPath, Time: utcTimestamp}
			if err := inotifyClient.Send(event); err != nil {
				// Stream is gone (typically a guest-agent reconnect). Return so
				// the caller can re-spawn against the new client instead of
				// looping silently with a dead stream.
				return fmt.Errorf("inotify stream closed: %w", err)
			}
		}
	}
}