mirror of
https://gitee.com/milvus-io/milvus.git
synced 2025-12-28 22:45:26 +08:00
621 lines
19 KiB
Go
621 lines
19 KiB
Go
// Licensed to the LF AI & Data foundation 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 proxy
|
|
|
|
import (
|
|
"encoding/base64"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"sync"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/samber/lo"
|
|
|
|
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
|
|
management "github.com/milvus-io/milvus/internal/http"
|
|
"github.com/milvus-io/milvus/internal/json"
|
|
"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
|
|
"github.com/milvus-io/milvus/pkg/v2/proto/internalpb"
|
|
"github.com/milvus-io/milvus/pkg/v2/proto/querypb"
|
|
"github.com/milvus-io/milvus/pkg/v2/util/commonpbutil"
|
|
"github.com/milvus-io/milvus/pkg/v2/util/merr"
|
|
)
|
|
|
|
// this file contains proxy management restful API handler
|
|
var mgrRouteRegisterOnce sync.Once
|
|
|
|
func RegisterMgrRoute(proxy *Proxy) {
|
|
mgrRouteRegisterOnce.Do(func() {
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteGcPause,
|
|
HandlerFunc: proxy.PauseDatacoordGC,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteGcResume,
|
|
HandlerFunc: proxy.ResumeDatacoordGC,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteListQueryNode,
|
|
HandlerFunc: proxy.ListQueryNode,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteGetQueryNodeDistribution,
|
|
HandlerFunc: proxy.GetQueryNodeDistribution,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteSuspendQueryCoordBalance,
|
|
HandlerFunc: proxy.SuspendQueryCoordBalance,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteResumeQueryCoordBalance,
|
|
HandlerFunc: proxy.ResumeQueryCoordBalance,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteSuspendQueryNode,
|
|
HandlerFunc: proxy.SuspendQueryNode,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteResumeQueryNode,
|
|
HandlerFunc: proxy.ResumeQueryNode,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteTransferSegment,
|
|
HandlerFunc: proxy.TransferSegment,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteTransferChannel,
|
|
HandlerFunc: proxy.TransferChannel,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteCheckQueryNodeDistribution,
|
|
HandlerFunc: proxy.CheckQueryNodeDistribution,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteQueryCoordBalanceStatus,
|
|
HandlerFunc: proxy.CheckQueryCoordBalanceStatus,
|
|
})
|
|
management.Register(&management.Handler{
|
|
Path: management.RouteBackupEZ,
|
|
HandlerFunc: proxy.BackupEZ,
|
|
})
|
|
})
|
|
}
|
|
|
|
// EncodeTicket encodes the ticket with token and collectionID
|
|
func EncodeTicket(token string, collectionID string) string {
|
|
if collectionID == "" {
|
|
collectionID = "-1"
|
|
}
|
|
m := map[string]string{
|
|
"token": token,
|
|
"collection_id": collectionID,
|
|
}
|
|
bytes, _ := json.Marshal(m)
|
|
ticket := base64.StdEncoding.EncodeToString(bytes)
|
|
return ticket
|
|
}
|
|
|
|
// DecodeTicket decodes the ticket to get token and collectionID
|
|
func DecodeTicket(ticket string) (string, string, error) {
|
|
bytes, err := base64.StdEncoding.DecodeString(ticket)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
m := make(map[string]string)
|
|
err = json.Unmarshal(bytes, &m)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
return m["token"], m["collection_id"], nil
|
|
}
|
|
|
|
func (node *Proxy) PauseDatacoordGC(w http.ResponseWriter, req *http.Request) {
|
|
pauseSeconds := req.URL.Query().Get("pause_seconds")
|
|
// generate ticket for request
|
|
token := uuid.New().String()
|
|
ticket := EncodeTicket(token, req.URL.Query().Get("collection_id"))
|
|
params := []*commonpb.KeyValuePair{
|
|
{Key: "duration", Value: pauseSeconds},
|
|
{Key: "ticket", Value: ticket},
|
|
}
|
|
if req.URL.Query().Has("collection_id") {
|
|
params = append(params, &commonpb.KeyValuePair{
|
|
Key: "collection_id",
|
|
Value: req.URL.Query().Get("collection_id"),
|
|
})
|
|
}
|
|
|
|
resp, err := node.mixCoord.GcControl(req.Context(), &datapb.GcControlRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
Command: datapb.GcCommand_Pause,
|
|
Params: params,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to pause garbage collection, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
if resp.GetErrorCode() != commonpb.ErrorCode_Success {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to pause garbage collection, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
fmt.Fprintf(w, `{"msg": "OK", "ticket": "%s"}`, ticket)
|
|
}
|
|
|
|
func (node *Proxy) ResumeDatacoordGC(w http.ResponseWriter, req *http.Request) {
|
|
ticket := req.URL.Query().Get("ticket")
|
|
var collectionID string
|
|
var err error
|
|
// allow empty ticket for backward compatibility
|
|
if ticket != "" {
|
|
_, collectionID, err = DecodeTicket(ticket)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to decode ticket, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
}
|
|
params := []*commonpb.KeyValuePair{
|
|
{Key: "ticket", Value: req.URL.Query().Get("ticket")},
|
|
{Key: "collection_id", Value: collectionID},
|
|
}
|
|
|
|
resp, err := node.mixCoord.GcControl(req.Context(), &datapb.GcControlRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
Command: datapb.GcCommand_Resume,
|
|
Params: params,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume garbage collection, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
if resp.GetErrorCode() != commonpb.ErrorCode_Success {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume garbage collection, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) ListQueryNode(w http.ResponseWriter, req *http.Request) {
|
|
resp, err := node.mixCoord.ListQueryNode(req.Context(), &querypb.ListQueryNodeRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to list query node, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp.GetStatus()) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to list query node, %s"}`, resp.GetStatus().GetReason())))
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusOK)
|
|
// skip marshal status to output
|
|
resp.Status = nil
|
|
bytes, err := json.Marshal(resp)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to list query node, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
w.Write(bytes)
|
|
}
|
|
|
|
func (node *Proxy) GetQueryNodeDistribution(w http.ResponseWriter, req *http.Request) {
|
|
err := req.ParseForm()
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
nodeID, err := strconv.ParseInt(req.FormValue("node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to get query node distribution, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
resp, err := node.mixCoord.GetQueryNodeDistribution(req.Context(), &querypb.GetQueryNodeDistributionRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
NodeID: nodeID,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to get query node distribution, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp.GetStatus()) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to get query node distribution, %s"}`, resp.GetStatus().GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
|
|
// Use string array for SealedSegmentIDs to prevent precision loss in JSON parsers.
|
|
// Large integers (int64) may be incorrectly rounded when parsed as double.
|
|
type distribution struct {
|
|
Channels []string `json:"channel_names"`
|
|
SealedSegmentIDs []string `json:"sealed_segmentIDs"`
|
|
}
|
|
|
|
dist := distribution{
|
|
Channels: resp.ChannelNames,
|
|
SealedSegmentIDs: lo.Map(resp.SealedSegmentIDs, func(id int64, _ int) string {
|
|
return strconv.FormatInt(id, 10)
|
|
}),
|
|
}
|
|
|
|
bytes, err := json.Marshal(dist)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to get query node distribution, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
w.Write(bytes)
|
|
}
|
|
|
|
func (node *Proxy) SuspendQueryCoordBalance(w http.ResponseWriter, req *http.Request) {
|
|
resp, err := node.mixCoord.SuspendBalance(req.Context(), &querypb.SuspendBalanceRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to suspend balance, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to suspend balance, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) ResumeQueryCoordBalance(w http.ResponseWriter, req *http.Request) {
|
|
resp, err := node.mixCoord.ResumeBalance(req.Context(), &querypb.ResumeBalanceRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume balance, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume balance, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) CheckQueryCoordBalanceStatus(w http.ResponseWriter, req *http.Request) {
|
|
resp, err := node.mixCoord.CheckBalanceStatus(req.Context(), &querypb.CheckBalanceStatusRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to check balance status, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp.GetStatus()) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to check balance status, %s"}`, resp.GetStatus().GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
balanceStatus := "suspended"
|
|
if resp.IsActive {
|
|
balanceStatus = "active"
|
|
}
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "OK", "status": "%v"}`, balanceStatus)))
|
|
}
|
|
|
|
func (node *Proxy) SuspendQueryNode(w http.ResponseWriter, req *http.Request) {
|
|
err := req.ParseForm()
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
nodeID, err := strconv.ParseInt(req.FormValue("node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to suspend node, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
resp, err := node.mixCoord.SuspendNode(req.Context(), &querypb.SuspendNodeRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
NodeID: nodeID,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to suspend node, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to suspend node, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) ResumeQueryNode(w http.ResponseWriter, req *http.Request) {
|
|
err := req.ParseForm()
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
nodeID, err := strconv.ParseInt(req.FormValue("node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume node, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
resp, err := node.mixCoord.ResumeNode(req.Context(), &querypb.ResumeNodeRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
NodeID: nodeID,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume node, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to resume node, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) TransferSegment(w http.ResponseWriter, req *http.Request) {
|
|
err := req.ParseForm()
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
request := &querypb.TransferSegmentRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
}
|
|
|
|
source, err := strconv.ParseInt(req.FormValue("source_node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": failed to transfer segment", %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.SourceNodeID = source
|
|
|
|
target := req.FormValue("target_node_id")
|
|
if len(target) == 0 {
|
|
request.ToAllNodes = true
|
|
} else {
|
|
value, err := strconv.ParseInt(target, 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.TargetNodeID = value
|
|
}
|
|
|
|
segmentID := req.FormValue("segment_id")
|
|
if len(segmentID) == 0 {
|
|
request.TransferAll = true
|
|
} else {
|
|
value, err := strconv.ParseInt(segmentID, 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.SegmentID = value
|
|
}
|
|
|
|
copyMode := req.FormValue("copy_mode")
|
|
if len(copyMode) == 0 {
|
|
request.CopyMode = true
|
|
} else {
|
|
value, err := strconv.ParseBool(copyMode)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.CopyMode = value
|
|
}
|
|
|
|
resp, err := node.mixCoord.TransferSegment(req.Context(), request)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer segment, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) TransferChannel(w http.ResponseWriter, req *http.Request) {
|
|
err := req.ParseForm()
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer channel, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
request := &querypb.TransferChannelRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
}
|
|
|
|
source, err := strconv.ParseInt(req.FormValue("source_node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": failed to transfer channel", %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.SourceNodeID = source
|
|
|
|
target := req.FormValue("target_node_id")
|
|
if len(target) == 0 {
|
|
request.ToAllNodes = true
|
|
} else {
|
|
value, err := strconv.ParseInt(target, 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer channel, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.TargetNodeID = value
|
|
}
|
|
|
|
channel := req.FormValue("channel_name")
|
|
if len(channel) == 0 {
|
|
request.TransferAll = true
|
|
} else {
|
|
request.ChannelName = channel
|
|
}
|
|
|
|
copyMode := req.FormValue("copy_mode")
|
|
if len(copyMode) == 0 {
|
|
request.CopyMode = false
|
|
} else {
|
|
value, err := strconv.ParseBool(copyMode)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer channel, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
request.CopyMode = value
|
|
}
|
|
|
|
resp, err := node.mixCoord.TransferChannel(req.Context(), request)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer channel, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to transfer channel, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) CheckQueryNodeDistribution(w http.ResponseWriter, req *http.Request) {
|
|
err := req.ParseForm()
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to check whether query node has same distribution, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
source, err := strconv.ParseInt(req.FormValue("source_node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": failed to check whether query node has same distribution", %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
target, err := strconv.ParseInt(req.FormValue("target_node_id"), 10, 64)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to check whether query node has same distribution, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
resp, err := node.mixCoord.CheckQueryNodeDistribution(req.Context(), &querypb.CheckQueryNodeDistributionRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
SourceNodeID: source,
|
|
TargetNodeID: target,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to check whether query node has same distribution, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to check whether query node has same distribution, %s"}`, resp.GetReason())))
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(`{"msg": "OK"}`))
|
|
}
|
|
|
|
func (node *Proxy) BackupEZ(w http.ResponseWriter, req *http.Request) {
|
|
dbName := req.URL.Query().Get("db_name")
|
|
if dbName == "" {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
w.Write([]byte(`{"msg": "db_name parameter is required"}`))
|
|
return
|
|
}
|
|
|
|
resp, err := node.mixCoord.BackupEzk(req.Context(), &internalpb.BackupEzkRequest{
|
|
Base: commonpbutil.NewMsgBase(),
|
|
DbName: dbName,
|
|
})
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to backup EZK, %s"}`, err.Error())))
|
|
return
|
|
}
|
|
|
|
if !merr.Ok(resp.GetStatus()) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "failed to backup EZK, %s"}`, resp.GetStatus().GetReason())))
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusOK)
|
|
w.Write([]byte(fmt.Sprintf(`{"msg": "OK", "ezk": "%s"}`, resp.Ezk)))
|
|
}
|