bitflicker64 commented on code in PR #366:
URL:
https://github.com/apache/hugegraph-computer/pull/366#discussion_r4353604243
##########
vermeer/apps/master/workers/worker_manager.go:
##########
@@ -233,6 +244,79 @@ func (wm *workerManager) CheckWorkerAlive(name string)
bool {
return true
}
+// SetWorkerConnection publishes a completed registration dial to readiness
readers.
+func (wm *workerManager) SetWorkerConnection(worker *WorkerClient, connection
*grpc.ClientConn) {
+ defer wm.Unlock(wm.Lock())
+ worker.Connection = connection
+}
+
+// Task streams are published only after their dispatch entry has been
installed.
+func (wm *workerManager) SetTaskStream(worker *WorkerClient, stream
TaskStream, ctx context.Context) {
+ defer wm.Unlock(wm.Lock())
+ if worker != nil && wm.workersByName[worker.Name] == worker {
+ worker.taskStreams[stream] = ctx
+ }
+}
+
+func (wm *workerManager) ClearTaskStream(worker *WorkerClient, stream
TaskStream, ctx context.Context) {
+ defer wm.Unlock(wm.Lock())
+ if worker != nil && wm.workersByName[worker.Name] == worker &&
worker.taskStreams[stream] == ctx {
+ worker.taskStreams[stream] = nil
+ }
+}
+
+func (wm *workerManager) IsWorkerReady(worker *WorkerClient) bool {
+ defer wm.Unlock(wm.Lock())
+ return wm.workerReadyLocked(worker)
+}
+
+func (wm *workerManager) workerReadyLocked(worker *WorkerClient) bool {
+ if worker == nil || wm.workersByName[worker.Name] != worker ||
+ worker.Connection == nil || worker.Connection.GetState() !=
connectivity.Ready {
+ return false
+ }
+ for _, ctx := range worker.taskStreams {
+ if ctx == nil || ctx.Err() != nil {
+ return false
+ }
+ }
+ return true
+}
+
+// HasAvailableWorkers follows existing allocation precedence without
allocating.
+// Every registered member needs both a ready reverse connection and live task
streams.
+func (wm *workerManager) HasAvailableWorkers(space string, graphNames
[]string) bool {
+ defer wm.Unlock(wm.Lock())
+ groups := make(map[string]bool)
+ group, explicit := wm.groupMapper[space]
+ if !explicit {
+ group = "$" // AddWorker normalizes ungrouped registered
workers to "$".
+ }
+ groups[group] = true
+ for _, graph := range graphNames {
Review Comment:
Minor: Availability ignores a group that an admin allocated to a graph that
is not loaded yet, while the scheduler uses that group for the graph's first
load.
Example: groupMapper {"current newgraph": "graph-group"}, one ready worker
in graph-group, no space-level group and no workers in "$".
ApplyGroup("current", "newgraph") returns graph-group and its member is ready,
but HasAvailableWorkers("current", nil) returns false, because graphNames only
comes from GraphMgr. AdminWorkerGroupGraphAllocHandler does not require the
graph to exist, so this layout is reachable, and a client that waits for this
flag would not start that first load.
The PR rules out prefix matching on purpose (the "other space with shared
prefix" test case), so the simplest fix is one README sentence next to the
endpoint saying such a group is not counted until the graph exists.
##########
vermeer/apps/master/bl/task_bl.go:
##########
@@ -344,7 +344,13 @@ func sendLoadingReq(task *structure.TaskInfo,
workerClients []*WorkerClient, app
}
task.Workers = append(task.Workers, &taskWorker)
//err := wc.loadServer.SendLoadReq(apply(wc))
- err := ServerMgr.LoadServer(wc.Name).SendLoadReq(apply(wc))
+ server := ServerMgr.LoadServer(wc.Name)
+ if server == nil {
+ err := fmt.Errorf("worker %s load task stream is
unavailable", wc.Name)
+ taskMgr.SetError(task, err.Error())
+ return err
Review Comment:
Minor: When a later worker has no load stream, this returns after earlier
workers in the loop already received LoadStep_Vertex, and nothing tells them to
stop.
workerClients comes from GroupWorkers (map order), so the worker without a
stream can come after one that was already sent the request. The dispatched
worker runs StartLoadGraph, but once the task is in Error its FetchLoadPart
calls get no partition (load_graph.go:370-372), its LoadTaskStatus reports are
ignored (load_task.go:81), and no LoadStep_Error reaches it, so its load task
and graph stay in memory. The SendLoadReq error branch below has the same gap.
Suggestion: before returning, send LoadStep_Error to the workers already
dispatched, skipping any whose LoadServer is nil. Checking every stream before
the first send is shorter but covers only this branch.
##########
vermeer/apps/master/services/http_worker_availability_test.go:
##########
@@ -0,0 +1,72 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with this
+work for additional information regarding copyright ownership. The ASF
+licenses this file to You under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+License for the specific language governing permissions and limitations
+under the License.
+*/
+
+package services
+
+import (
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+ "vermeer/apps/common"
+ "vermeer/apps/structure"
+
+ "github.com/gin-gonic/gin"
+ "github.com/prometheus/client_golang/prometheus"
+)
+
+func TestWorkerAvailabilityUsesAuthenticatedSpace(t *testing.T) {
+ credential, err := structure.NewCredential("space-user", "own-space")
+ if err != nil {
+ t.Fatal(err)
+ }
+ previousGraphs := common.PrometheusMetrics.GraphCnt
+ common.PrometheusMetrics.GraphCnt = prometheus.NewGaugeVec(
+ prometheus.GaugeOpts{Name: "test_availability_graphs"}, nil)
+ GraphMgr.Init()
+ t.Cleanup(func() { GraphMgr.Init(); common.PrometheusMetrics.GraphCnt =
previousGraphs })
+ for _, space := range []string{"own-space", "another-space"} {
+ if err := GraphMgr.AddGraph(GraphMgr.CreateGraph(space,
space+"-graph")); err != nil {
+ t.Fatal(err)
+ }
+ }
+ router := gin.New()
+ var checked string
+ router.GET("/workers/availability", func(ctx *gin.Context) {
Review Comment:
Minor: This test registers its own route and sets the credential itself, so
nothing checks that /workers/availability sits behind the auth filter, which is
the access rule the README states for this endpoint.
Through SetRouters(sen, auth.TokenFilter), the endpoint returns 401 without
a token on both /api/v1/workers/availability and /workers/availability, and 200
with a signed Bearer token. So the wiring is correct today, but removing
authFilters from the new regVerAPI call in router.go would still pass this test.
Suggestion: add a case that goes through SetRouters with auth.TokenFilter
and expects 401 without a token. Sentinel keeps its gin engine unexported, so
the test needs a small accessor or has to start the server on a loopback port.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
