frostfs-node/pkg/services/control/server/policy_engine.go

207 lines
6.9 KiB
Go
Raw Normal View History

package control
import (
"context"
"errors"
"fmt"
"strings"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/control"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
apechain "git.frostfs.info/TrueCloudLab/policy-engine/pkg/chain"
engine "git.frostfs.info/TrueCloudLab/policy-engine/pkg/engine"
nativeschema "git.frostfs.info/TrueCloudLab/policy-engine/schema/native"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// extractCID extracts CID from the schema's pattern.
// TODO (aarifullin): This is temporary solution should be replaced by
// resource name validation.
func extractCID(resource string) (cid.ID, error) {
var cidStr string
// Sscanf requires to make tokens delimited by spaces.
pattern := strings.Replace(nativeschema.ResourceFormatRootContainerObjects, "/", " ", -1)
resource = strings.Replace(resource, "/", " ", -1)
if _, err := fmt.Sscanf(resource, pattern, &cidStr); err != nil {
err = fmt.Errorf("could not parse the target name '%s' to CID: %w", resource, err)
return cid.ID{}, err
}
var cid cid.ID
err := cid.DecodeString(cidStr)
return cid, err
}
func (s *Server) AddChainLocalOverride(_ context.Context, req *control.AddChainLocalOverrideRequest) (*control.AddChainLocalOverrideResponse, error) {
if err := s.isValidRequest(req); err != nil {
return nil, status.Error(codes.PermissionDenied, err.Error())
}
target := req.GetBody().GetTarget()
if target.Type != control.ChainTarget_CONTAINER {
return nil, status.Error(codes.Internal, fmt.Errorf("target type is not supported: %s", target.Type.String()).Error())
}
cid, err := extractCID(target.GetName())
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
var chain apechain.Chain
if err = chain.DecodeBytes(req.GetBody().GetChain()); err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
src, err := s.apeChainSrc.GetChainSource(cid)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
s.apeChainCounter.Add(1)
// TODO (aarifullin): the such chain id is not well-designed yet.
if chain.ID == "" {
chain.ID = apechain.ID(fmt.Sprintf("%s:%d", apechain.Ingress, s.apeChainCounter.Load()))
}
resource := fmt.Sprintf(nativeschema.ResourceFormatRootContainerObjects, cid.EncodeToString())
if _, err = src.LocalStorage().AddOverride(apechain.Ingress, engine.ContainerTarget(resource), &chain); err != nil {
return nil, status.Error(getCodeByLocalStorageErr(err), err.Error())
}
resp := &control.AddChainLocalOverrideResponse{
Body: &control.AddChainLocalOverrideResponse_Body{
ChainId: string(chain.ID),
},
}
err = SignMessage(s.key, resp)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
return resp, nil
}
func (s *Server) GetChainLocalOverride(_ context.Context, req *control.GetChainLocalOverrideRequest) (*control.GetChainLocalOverrideResponse, error) {
if err := s.isValidRequest(req); err != nil {
return nil, status.Error(codes.PermissionDenied, err.Error())
}
target := req.GetBody().GetTarget()
if target.Type != control.ChainTarget_CONTAINER {
return nil, status.Error(codes.Internal, fmt.Errorf("target type is not supported: %s", target.Type.String()).Error())
}
cid, err := extractCID(target.GetName())
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
src, err := s.apeChainSrc.GetChainSource(cid)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
resource := fmt.Sprintf(nativeschema.ResourceFormatRootContainerObjects, cid.EncodeToString())
chain, err := src.LocalStorage().GetOverride(apechain.Ingress, engine.ContainerTarget(resource), apechain.ID(req.GetBody().GetChainId()))
if err != nil {
return nil, status.Error(getCodeByLocalStorageErr(err), err.Error())
}
resp := &control.GetChainLocalOverrideResponse{
Body: &control.GetChainLocalOverrideResponse_Body{
Chain: chain.Bytes(),
},
}
err = SignMessage(s.key, resp)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
return resp, nil
}
func (s *Server) ListChainLocalOverrides(_ context.Context, req *control.ListChainLocalOverridesRequest) (*control.ListChainLocalOverridesResponse, error) {
if err := s.isValidRequest(req); err != nil {
return nil, status.Error(codes.PermissionDenied, err.Error())
}
target := req.GetBody().GetTarget()
if target.Type != control.ChainTarget_CONTAINER {
return nil, status.Error(codes.Internal, fmt.Errorf("target type is not supported: %s", target.Type.String()).Error())
}
cid, err := extractCID(target.GetName())
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
src, err := s.apeChainSrc.GetChainSource(cid)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
resource := fmt.Sprintf(nativeschema.ResourceFormatRootContainerObjects, cid.EncodeToString())
chains, err := src.LocalStorage().ListOverrides(apechain.Ingress, engine.ContainerTarget(resource))
if err != nil {
return nil, status.Error(getCodeByLocalStorageErr(err), err.Error())
}
serializedChains := make([][]byte, 0, len(chains))
for _, chain := range chains {
serializedChains = append(serializedChains, chain.Bytes())
}
resp := &control.ListChainLocalOverridesResponse{
Body: &control.ListChainLocalOverridesResponse_Body{
Chains: serializedChains,
},
}
err = SignMessage(s.key, resp)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
return resp, nil
}
func (s *Server) RemoveChainLocalOverride(_ context.Context, req *control.RemoveChainLocalOverrideRequest) (*control.RemoveChainLocalOverrideResponse, error) {
if err := s.isValidRequest(req); err != nil {
return nil, status.Error(codes.PermissionDenied, err.Error())
}
target := req.GetBody().GetTarget()
if target.Type != control.ChainTarget_CONTAINER {
return nil, status.Error(codes.Internal, fmt.Errorf("target type is not supported: %s", target.Type.String()).Error())
}
cid, err := extractCID(target.GetName())
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
src, err := s.apeChainSrc.GetChainSource(cid)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
resource := fmt.Sprintf(nativeschema.ResourceFormatRootContainerObjects, cid.EncodeToString())
if err = src.LocalStorage().RemoveOverride(apechain.Ingress, engine.ContainerTarget(resource), apechain.ID(req.GetBody().GetChainId())); err != nil {
return nil, status.Error(getCodeByLocalStorageErr(err), err.Error())
}
resp := &control.RemoveChainLocalOverrideResponse{
Body: &control.RemoveChainLocalOverrideResponse_Body{
Removed: true,
},
}
err = SignMessage(s.key, resp)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
return resp, nil
}
func getCodeByLocalStorageErr(err error) codes.Code {
if errors.Is(err, engine.ErrChainNotFound) {
return codes.NotFound
}
return codes.Internal
}