Nit, consider using an interface instead of implementation.

Yup, this is required for testing, like mentioned in the struct comment just above. If it's fine with you, I'll add an interface in pkg/node/manager/manager.go.

The interface could be defined here, not in pkg/node/manager/manager.go, this allows you to not need to import pkg/node/manager in this package.

I've added the interface with this commit faae24149afbb9da49688945123c09a79716f903 already and implemented tests of the service.
I think that importing the interface from pkg/node/manager is not so bad because it's otherwise really hard to understand who implements it if you're not familiar enough with cilium's codebase.

@rolinh there are tools built for that purpose [0] so there's no point in keeping the interfaces in the same place as 1 of the implementations. What happens if in the future we will have a different node manager that implements the same interface? Where will the interface live?

[0] https://docs.google.com/document/d/1_Y9xCEMj5S-7rv2ooHpZNH15JgRT5iM742gJkw5LtmQ/edit#heading=h.f6hyappclfor

I don't mind moving the interface to pkg/hubble/peer if you have strong feelings about it; I don't think it matters much.

The interface could be defined here, not in pkg/node/manager/manager.go, this allows you to not need to import pkg/node/manager in this package.

BTW, by moving the interface to pkg/hubble/peer I could save on importing pkg/node/manager but then I'll have to import pkg/datapath as it's required for the interface definition. I'm not "saving" on import anyway.

stream.Send() can block (it does I/O). Which means the unbuffered h.C channel causes the NodeManager to be blocked as well in h.Node{Add,Update,Delete}.

Why isn't a context.Context part of the Send parameters?

Because the ServerStream interface doesn't use a context.Context. SendMsg() is used by Send(), as can be seen in the generated code in peer.pb.go:

type Peer_NotifyServer interface {
	Send(*ChangeNotification) error
	grpc.ServerStream
}

type peerNotifyServer struct {
	grpc.ServerStream
}

func (x *peerNotifyServer) Send(m *ChangeNotification) error {
	return x.ServerStream.SendMsg(m)
}

The interface defines a context but it's global to ServerStream.

@gandro I have added a test that simulates a blocked send (TestService_NotifyWithBlockedSend) and it failed (which is good as it tests what we want).

To fix the problem and have the test pass, I've added small buffer (of size 10 by default but this is configurable through service options) that, when full, triggers the server to stop steaming to the client.
This eventually pushes the problem back to the client which will have to call Notify() again in such a case. I don't think it's a problem overall because upon calling Notify(), the client receives the list of all peers before receiving continuous updates so maintaining a local cache client-side is easy.

Please, have a look again at the Notify() implementation and tell me if you think this way of handling the problem is OK.

Awesome, yes, the code itself looks good! 🎉 We have to get a bit of experience with this, because I honestly have no clue how big the buffer size should be.

We have to get a bit of experience with this, because I honestly have no clue how big the buffer size should be.

Neither do I. I assumed that since the connection is over a local unix domain socket, a buffer of 10 should be sufficient. I also don't want to make it unnecessarily large because if the caller can't keep up then there's probably a problem elsewhere.

The interface could be defined here, not in pkg/node/manager/manager.go, this allows you to not need to import pkg/node/manager in this package.

Why isn't a context.Context part of the Send parameters?

Awesome, yes, the code itself looks good! 🎉 We have to get a bit of experience with this, because I honestly have no clue how big the buffer size should be.