MCPcopy Create free account
hub / github.com/rilldata/rill / WatchResources

Method WatchResources

runtime/server/controller.go:93–157  ·  view source on GitHub ↗

WatchResources implements runtimev1.RuntimeServiceServer

(req *runtimev1.WatchResourcesRequest, ss runtimev1.RuntimeService_WatchResourcesServer)

Source from the content-addressed store, hash-verified

91
92// WatchResources implements runtimev1.RuntimeServiceServer
93func (s *Server) WatchResources(req *runtimev1.WatchResourcesRequest, ss runtimev1.RuntimeService_WatchResourcesServer) error {
94 observability.AddRequestAttributes(ss.Context(),
95 attribute.String("args.instance_id", req.InstanceId),
96 attribute.String("args.kind", req.Kind),
97 )
98
99 claims := auth.GetClaims(ss.Context(), req.InstanceId)
100 if !claims.Can(runtime.ReadObjects) {
101 return ErrForbidden
102 }
103
104 ctrl, err := s.runtime.Controller(ss.Context(), req.InstanceId)
105 if err != nil {
106 return err
107 }
108
109 if req.Replay {
110 rs, err := ctrl.List(ss.Context(), req.Kind, "", false)
111 if err != nil {
112 return err
113 }
114
115 for _, r := range rs {
116 r, access, err := s.runtime.ApplySecurityPolicy(ss.Context(), req.InstanceId, claims, r)
117 if err != nil {
118 return mapGRPCErrorWithFallback(err, codes.InvalidArgument)
119 }
120 if !access {
121 continue
122 }
123
124 err = ss.Send(&runtimev1.WatchResourcesResponse{
125 Event: runtimev1.ResourceEvent_RESOURCE_EVENT_WRITE,
126 Resource: r,
127 })
128 if err != nil {
129 return err
130 }
131 }
132 }
133
134 return ctrl.Subscribe(ss.Context(), func(e runtimev1.ResourceEvent, n *runtimev1.ResourceName, r *runtimev1.Resource) {
135 if r != nil { // r is nil for deletion events
136 var access bool
137 var err error
138 r, access, err = s.runtime.ApplySecurityPolicy(ss.Context(), req.InstanceId, claims, r)
139 if err != nil {
140 s.logger.Info("failed to apply security policy", zap.String("name", n.Name), zap.Error(err))
141 return
142 }
143 if !access {
144 return
145 }
146 }
147
148 err = ss.Send(&runtimev1.WatchResourcesResponse{
149 Event: e,
150 Name: n,

Callers 1

SSEHandlerMethod · 0.95

Calls 13

AddRequestAttributesFunction · 0.92
GetClaimsFunction · 0.92
mapGRPCErrorWithFallbackFunction · 0.85
CanMethod · 0.80
ControllerMethod · 0.80
ListMethod · 0.80
ApplySecurityPolicyMethod · 0.80
InfoMethod · 0.80
ContextMethod · 0.65
StringMethod · 0.65
SendMethod · 0.65
SubscribeMethod · 0.45

Tested by

no test coverage detected