JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/internalpb: move datarepair.pb to internal

Browse Source 
Change-Id: If901d9ff4e5ee6715b963eeeb46513a602a44b3d
This commit is contained in:
Egon Elbre 2020-10-30 12:41:22 +02:00
parent 8f26f66da0
commit 004e610d0f
13 changed files with 238 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
satellite/internalpb/datarepair.pb.go Normal file
View File

@ -0,0 +1,104 @@
// Code generated by protoc-gen-gogo. DO NOT EDIT.
// source: datarepair.proto
package internalpb
import (
fmt "fmt"
math "math"
time "time"
proto "github.com/gogo/protobuf/proto"
)
// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
var _ = math.Inf
var _ = time.Kitchen
// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
// InjuredSegment is the queue item used for the data repair queue.
type InjuredSegment struct {
Path []byte `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"`
LostPieces []int32 `protobuf:"varint,2,rep,packed,name=lost_pieces,json=lostPieces,proto3" json:"lost_pieces,omitempty"`
InsertedTime time.Time `protobuf:"bytes,3,opt,name=inserted_time,json=insertedTime,proto3,stdtime" json:"inserted_time"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *InjuredSegment) Reset() { *m = InjuredSegment{} }
func (m *InjuredSegment) String() string { return proto.CompactTextString(m) }
func (*InjuredSegment) ProtoMessage() {}
func (*InjuredSegment) Descriptor() ([]byte, []int) {
return fileDescriptor_b1b08e6fe9398aa6, []int{0}
}
func (m *InjuredSegment) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_InjuredSegment.Unmarshal(m, b)
}
func (m *InjuredSegment) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
return xxx_messageInfo_InjuredSegment.Marshal(b, m, deterministic)
}
func (m *InjuredSegment) XXX_Merge(src proto.Message) {
xxx_messageInfo_InjuredSegment.Merge(m, src)
}
func (m *InjuredSegment) XXX_Size() int {
return xxx_messageInfo_InjuredSegment.Size(m)
}
func (m *InjuredSegment) XXX_DiscardUnknown() {
xxx_messageInfo_InjuredSegment.DiscardUnknown(m)
}
var xxx_messageInfo_InjuredSegment proto.InternalMessageInfo
func (m *InjuredSegment) GetPath() []byte {
if m != nil {
return m.Path
}
return nil
}
func (m *InjuredSegment) GetLostPieces() []int32 {
if m != nil {
return m.LostPieces
}
return nil
}
func (m *InjuredSegment) GetInsertedTime() time.Time {
if m != nil {
return m.InsertedTime
}
return time.Time{}
}
func init() {
proto.RegisterType((*InjuredSegment)(nil), "repair.InjuredSegment")
}
func init() { proto.RegisterFile("datarepair.proto", fileDescriptor_b1b08e6fe9398aa6) }
var fileDescriptor_b1b08e6fe9398aa6 = []byte{
// 226 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x34, 0x8e, 0x41, 0x4e, 0xc3, 0x30,
0x10, 0x45, 0x31, 0x85, 0x0a, 0xb9, 0x05, 0x21, 0xaf, 0xa2, 0x6c, 0x12, 0x81, 0x90, 0xb2, 0xb2,
0x25, 0xb8, 0x41, 0x77, 0xdd, 0xa1, 0xc0, 0x8a, 0x4d, 0xe5, 0x90, 0xc1, 0xb8, 0x72, 0x3c, 0x96,
0x3d, 0xbd, 0x47, 0x8f, 0xc5, 0x29, 0xe0, 0x2a, 0x28, 0xb6, 0xb2, 0xfb, 0xff, 0xd9, 0x33, 0xf3,
0xf8, 0xfd, 0xa8, 0x49, 0x47, 0x08, 0xda, 0x46, 0x19, 0x22, 0x12, 0x8a, 0x75, 0x69, 0x35, 0x37,
0x68, 0xb0, 0xb0, 0xba, 0x31, 0x88, 0xc6, 0x81, 0xca, 0x6d, 0x38, 0x7d, 0x29, 0xb2, 0x13, 0x24,
0xd2, 0x53, 0x28, 0x1f, 0x1e, 0xce, 0x8c, 0xdf, 0xed, 0xfd, 0xf1, 0x14, 0x61, 0x7c, 0x03, 0x33,
0x81, 0x27, 0x21, 0xf8, 0x55, 0xd0, 0xf4, 0x5d, 0xb1, 0x96, 0x75, 0xdb, 0x3e, 0x67, 0xd1, 0xf0,
0x8d, 0xc3, 0x44, 0x87, 0x60, 0xe1, 0x13, 0x52, 0x75, 0xd9, 0xae, 0xba, 0xeb, 0x9e, 0xcf, 0xe8,
0x35, 0x13, 0xb1, 0xe7, 0xb7, 0xd6, 0x27, 0x88, 0x04, 0xe3, 0x61, 0xbe, 0x51, 0xad, 0x5a, 0xd6,
0x6d, 0x9e, 0x6b, 0x59, 0x04, 0xe4, 0x22, 0x20, 0xdf, 0x17, 0x81, 0xdd, 0xcd, 0xcf, 0x6f, 0x73,
0x71, 0xfe, 0x6b, 0x58, 0xbf, 0x5d, 0x46, 0xe7, 0xc7, 0xdd, 0xd3, 0xc7, 0x63, 0x22, 0x8c, 0x47,
0x69, 0x51, 0xe5, 0xa0, 0x92, 0x26, 0x70, 0xce, 0x12, 0x28, 0xeb, 0x09, 0xa2, 0xd7, 0x2e, 0x0c,
0xc3, 0x3a, 0xaf, 0x7c, 0xf9, 0x0f, 0x00, 0x00, 0xff, 0xff, 0x99, 0xad, 0xb3, 0xd4, 0x09, 0x01,
0x00, 0x00,
}

17
satellite/internalpb/datarepair.proto Normal file
View File

@ -0,0 +1,17 @@
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
syntax = "proto3";
option go_package = "storj.io/storj/satellite/internalpb";
import "gogo.proto";
import "google/protobuf/timestamp.proto";
package repair;
// InjuredSegment is the queue item used for the data repair queue.
message InjuredSegment {
bytes path = 1;
repeated int32 lost_pieces = 2;
google.protobuf.Timestamp inserted_time = 3 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
}

7
satellite/internalpb/doc.go Normal file
View File

@ -0,0 +1,7 @@
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
//go:generate go run gen.go
// Package internalpb contains proto definitions for satellite internal tools.
package internalpb

8
satellite/internalpb/gen.go
View File

@ -3,8 +3,6 @@
// +build ignore
//go:generate go run gen.go
package main
import (
@ -18,7 +16,7 @@ import (
)
var (
mainpkg = flag.String("pkg", "storj.io/storj/internalpb", "main package name")
mainpkg = flag.String("pkg", "storj.io/storj/satellite/internalpb", "main package name")
protoc = flag.String("protoc", "protoc", "protoc compiler")
)
@ -67,10 +65,10 @@ func main() {
commonPb := os.Getenv("STORJ_COMMON_PB")
if commonPb == "" {
commonPb = "../../common/pb"
commonPb = "../../../common/pb"
}
overrideImports := ",Mgoogle/protobuf/timestamp.proto=storj.io/storj/internalpb"
overrideImports := ",Mgoogle/protobuf/timestamp.proto=storj.io/storj/satellite/internalpb"
args := []string{
"--lint_out=.",
"--drpc_out=plugins=drpc,paths=source_relative" + overrideImports + ":.",

64
satellite/internalpb/metainfo_sat.pb.go
View File

@ -207,36 +207,36 @@ func init() {
func init() { proto.RegisterFile("metainfo_sat.proto", fileDescriptor_47c60bd892d94aaf) }
var fileDescriptor_47c60bd892d94aaf = []byte{
// 492 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x52, 0xcb, 0x8e, 0xd3, 0x40,
0x10, 0xc4, 0x1b, 0xf2, 0x9a, 0xbc, 0xa4, 0x59, 0x40, 0x26, 0x11, 0x4a, 0xb4, 0x12, 0x52, 0x4e,
0xb6, 0xb4, 0x7b, 0xe4, 0x44, 0x94, 0x8b, 0x25, 0x1e, 0x8b, 0xc3, 0x89, 0x8b, 0x35, 0xf6, 0xf4,
0x3a, 0x03, 0xf6, 0x8c, 0x35, 0xd3, 0x46, 0xbb, 0x7f, 0xc0, 0x91, 0xcf, 0xe2, 0x1b, 0x38, 0x2c,
0x9f, 0xc1, 0x15, 0x79, 0x1c, 0xc7, 0x91, 0xd0, 0x1e, 0xe0, 0xe6, 0xaa, 0xae, 0xea, 0xe9, 0xee,
0x32, 0xa1, 0x39, 0x20, 0x13, 0xf2, 0x46, 0x45, 0x86, 0xa1, 0x57, 0x68, 0x85, 0x8a, 0x0e, 0x1a,
0x6e, 0x4e, 0x52, 0x95, 0xaa, 0x9a, 0x9d, 0x2f, 0x53, 0xa5, 0xd2, 0x0c, 0x7c, 0x8b, 0xe2, 0xf2,
0xc6, 0x47, 0x91, 0x83, 0x41, 0x96, 0x17, 0x07, 0xc1, 0xac, 0x50, 0x42, 0x22, 0x68, 0x1e, 0x1f,
0x88, 0x69, 0xd3, 0xa7, 0xc6, 0x17, 0xdf, 0x3a, 0x64, 0xb0, 0x43, 0x0d, 0x2c, 0x0f, 0xb6, 0xf4,
0x19, 0xe9, 0xc5, 0x65, 0xf2, 0x05, 0xd0, 0x75, 0x56, 0xce, 0x7a, 0x1c, 0x1e, 0x10, 0x7d, 0x49,
0xa6, 0x20, 0x13, 0x7d, 0x57, 0x20, 0xf0, 0xa8, 0x60, 0xb8, 0x77, 0xcf, 0x6c, 0x7d, 0x72, 0x64,
0xaf, 0x19, 0xee, 0xa9, 0x4b, 0xfa, 0x5f, 0x41, 0x1b, 0xa1, 0xa4, 0xdb, 0x59, 0x39, 0xeb, 0x6e,
0xd8, 0x40, 0xfa, 0x8a, 0x10, 0x0d, 0xbc, 0x94, 0x9c, 0xc9, 0xe4, 0xce, 0x7d, 0xbc, 0x72, 0xd6,
0xa3, 0xcb, 0x85, 0xd7, 0xce, 0x16, 0x1e, 0x8b, 0xbb, 0x64, 0x0f, 0x39, 0x84, 0x27, 0x72, 0x1a,
0x90, 0x49, 0xa2, 0x81, 0xa1, 0x50, 0x32, 0xe2, 0x0c, 0xc1, 0xed, 0x5a, 0xff, 0xdc, 0xab, 0x97,
0xf7, 0x9a, 0xe5, 0xbd, 0x8f, 0xcd, 0xf2, 0x9b, 0xc1, 0x8f, 0xfb, 0xe5, 0xa3, 0xef, 0xbf, 0x96,
0x4e, 0x38, 0x6e, 0xac, 0x5b, 0x86, 0x40, 0xdf, 0x92, 0x19, 0xdc, 0x16, 0x42, 0x9f, 0x34, 0xeb,
0xfd, 0x43, 0xb3, 0x69, 0x6b, 0xb6, 0xed, 0x7c, 0x72, 0x6e, 0x18, 0x42, 0x96, 0x09, 0x84, 0xc8,
0x88, 0x54, 0x32, 0x2c, 0x35, 0xb8, 0x43, 0x7b, 0x1c, 0x7a, 0x2c, 0xed, 0x9a, 0x0a, 0x5d, 0x90,
0xa1, 0xb1, 0xc7, 0x8e, 0x04, 0x77, 0x89, 0x95, 0x0d, 0x6a, 0x22, 0xe0, 0x17, 0xbf, 0xcf, 0xc8,
0x70, 0x07, 0x69, 0x0e, 0x12, 0x83, 0x2d, 0xf5, 0x4f, 0xa5, 0x8e, 0x1d, 0x92, 0x7a, 0xc7, 0xf0,
0x9a, 0xc8, 0x5a, 0x3b, 0x5d, 0x92, 0x51, 0xc1, 0x34, 0x46, 0xb2, 0xcc, 0x63, 0xd0, 0x36, 0xa1,
0x6e, 0x48, 0x2a, 0xea, 0x9d, 0x65, 0xe8, 0x13, 0xd2, 0x15, 0x92, 0xc3, 0xed, 0x21, 0x9c, 0x1a,
0xd0, 0x2b, 0x32, 0xd1, 0x4a, 0x61, 0x54, 0x08, 0x48, 0xa0, 0x7a, 0xab, 0xba, 0xee, 0x78, 0x33,
0xab, 0x96, 0xfe, 0x79, 0xbf, 0xec, 0x5f, 0x57, 0x7c, 0xb0, 0x0d, 0x47, 0x95, 0xaa, 0x06, 0x9c,
0x7e, 0x20, 0x4f, 0x95, 0x16, 0xa9, 0x90, 0x2c, 0x8b, 0x94, 0xe6, 0xa0, 0xa3, 0x4c, 0xe4, 0x02,
0x8d, 0xdb, 0x5b, 0x75, 0xd6, 0xa3, 0xcb, 0x17, 0xed, 0xa0, 0xaf, 0x39, 0xd7, 0x60, 0x0c, 0xf0,
0xf7, 0x95, 0xec, 0x4d, 0xa5, 0x0a, 0xcf, 0x1b, 0x6f, 0xcb, 0x99, 0xbf, 0x53, 0xee, 0xff, 0x77,
0xca, 0x0f, 0xc4, 0x32, 0x78, 0x28, 0x96, 0xcd, 0xe2, 0xd3, 0x73, 0x83, 0x4a, 0x7f, 0xf6, 0x84,
0xf2, 0xed, 0x87, 0x6f, 0x7f, 0x4c, 0xc9, 0xb2, 0x22, 0x8e, 0x7b, 0xf6, 0xe5, 0xab, 0x3f, 0x01,
0x00, 0x00, 0xff, 0xff, 0xcf, 0xd5, 0x29, 0xca, 0x96, 0x03, 0x00, 0x00,
// 495 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x52, 0x4d, 0x6f, 0xd3, 0x40,
0x10, 0xc5, 0x0d, 0xf9, 0xda, 0x7c, 0x49, 0x5b, 0x40, 0x56, 0x2a, 0x94, 0xa8, 0xa8, 0x52, 0x4e,
0xb6, 0xd4, 0x1e, 0x39, 0x11, 0xe5, 0x12, 0x89, 0x8f, 0xe2, 0x70, 0xe2, 0x62, 0xad, 0xbd, 0x53,
0x67, 0xc1, 0xde, 0xb5, 0x76, 0xc7, 0xa8, 0xfd, 0x07, 0x1c, 0xf9, 0x59, 0xfc, 0x06, 0x0e, 0xe5,
0x67, 0x70, 0x45, 0x5e, 0xc7, 0x76, 0x24, 0xd4, 0x03, 0xbd, 0xf9, 0xbd, 0x79, 0x6f, 0x3c, 0x33,
0x6f, 0x09, 0xcd, 0x00, 0x99, 0x90, 0x37, 0x2a, 0x34, 0x0c, 0xbd, 0x5c, 0x2b, 0x54, 0x74, 0x50,
0x73, 0x73, 0x92, 0xa8, 0x44, 0x55, 0xec, 0x7c, 0x91, 0x28, 0x95, 0xa4, 0xe0, 0x5b, 0x14, 0x15,
0x37, 0x3e, 0x8a, 0x0c, 0x0c, 0xb2, 0x2c, 0x3f, 0x08, 0x66, 0xb9, 0x12, 0x12, 0x41, 0xf3, 0xe8,
0x40, 0x4c, 0xeb, 0x3e, 0x15, 0x3e, 0xff, 0xde, 0x21, 0x83, 0x1d, 0x6a, 0x60, 0xd9, 0x76, 0x43,
0x5f, 0x90, 0x5e, 0x54, 0xc4, 0x5f, 0x01, 0x5d, 0x67, 0xe9, 0xac, 0xc6, 0xc1, 0x01, 0xd1, 0x0b,
0x32, 0x05, 0x19, 0xeb, 0xbb, 0x1c, 0x81, 0x87, 0x39, 0xc3, 0xbd, 0x7b, 0x62, 0xeb, 0x93, 0x86,
0xbd, 0x66, 0xb8, 0xa7, 0x2e, 0xe9, 0x7f, 0x03, 0x6d, 0x84, 0x92, 0x6e, 0x67, 0xe9, 0xac, 0xba,
0x41, 0x0d, 0xe9, 0x6b, 0x42, 0x34, 0xf0, 0x42, 0x72, 0x26, 0xe3, 0x3b, 0xf7, 0xe9, 0xd2, 0x59,
0x8d, 0x2e, 0xcf, 0xbc, 0x76, 0xb6, 0xa0, 0x29, 0xee, 0xe2, 0x3d, 0x64, 0x10, 0x1c, 0xc9, 0xe9,
0x96, 0x4c, 0x62, 0x0d, 0x0c, 0x85, 0x92, 0x21, 0x67, 0x08, 0x6e, 0xd7, 0xfa, 0xe7, 0x5e, 0xb5,
0xbc, 0x57, 0x2f, 0xef, 0x7d, 0xaa, 0x97, 0x5f, 0x0f, 0x7e, 0xde, 0x2f, 0x9e, 0xfc, 0xf8, 0xbd,
0x70, 0x82, 0x71, 0x6d, 0xdd, 0x30, 0x04, 0xfa, 0x8e, 0xcc, 0xe0, 0x36, 0x17, 0xfa, 0xa8, 0x59,
0xef, 0x3f, 0x9a, 0x4d, 0x5b, 0xb3, 0x6d, 0xe7, 0x93, 0x53, 0xc3, 0x10, 0xd2, 0x54, 0x20, 0x84,
0x46, 0x24, 0x92, 0x61, 0xa1, 0xc1, 0x1d, 0xda, 0xe3, 0xd0, 0xa6, 0xb4, 0xab, 0x2b, 0xf4, 0x8c,
0x0c, 0x8d, 0x3d, 0x76, 0x28, 0xb8, 0x4b, 0xac, 0x6c, 0x50, 0x11, 0x5b, 0x7e, 0xfe, 0xe7, 0x84,
0x0c, 0x77, 0x90, 0x64, 0x20, 0x71, 0xbb, 0xa1, 0xfe, 0xb1, 0xd4, 0xb1, 0x43, 0x52, 0xaf, 0x09,
0xaf, 0x8e, 0xac, 0xb5, 0xd3, 0x05, 0x19, 0xe5, 0x4c, 0x63, 0x28, 0x8b, 0x2c, 0x02, 0x6d, 0x13,
0xea, 0x06, 0xa4, 0xa4, 0xde, 0x5b, 0x86, 0x3e, 0x23, 0x5d, 0x21, 0x39, 0xdc, 0x1e, 0xc2, 0xa9,
0x00, 0xbd, 0x22, 0x13, 0xad, 0x14, 0x86, 0xb9, 0x80, 0x18, 0xca, 0x7f, 0x95, 0xd7, 0x1d, 0xaf,
0x67, 0xe5, 0xd2, 0xbf, 0xee, 0x17, 0xfd, 0xeb, 0x92, 0xdf, 0x6e, 0x82, 0x51, 0xa9, 0xaa, 0x00,
0xa7, 0x1f, 0xc9, 0x73, 0xa5, 0x45, 0x22, 0x24, 0x4b, 0x43, 0xa5, 0x39, 0xe8, 0x30, 0x15, 0x99,
0x40, 0xe3, 0xf6, 0x96, 0x9d, 0xd5, 0xe8, 0xf2, 0x65, 0x3b, 0xe8, 0x1b, 0xce, 0x35, 0x18, 0x03,
0xfc, 0x43, 0x29, 0x7b, 0x5b, 0xaa, 0x82, 0xd3, 0xda, 0xdb, 0x72, 0xe6, 0xdf, 0x94, 0xfb, 0x8f,
0x4e, 0xf9, 0x81, 0x58, 0x06, 0x0f, 0xc5, 0xb2, 0xbe, 0xf8, 0xfc, 0xca, 0xa0, 0xd2, 0x5f, 0x3c,
0xa1, 0x7c, 0xfb, 0xe1, 0x37, 0x22, 0xdf, 0x3e, 0x51, 0xc9, 0xd2, 0x3c, 0x8a, 0x7a, 0x76, 0x86,
0xab, 0xbf, 0x01, 0x00, 0x00, 0xff, 0xff, 0x5e, 0x94, 0xf1, 0x28, 0xa0, 0x03, 0x00, 0x00,
}

2
satellite/internalpb/metainfo_sat.proto
View File

@ -2,7 +2,7 @@
// See LICENSE for copying information.
syntax = "proto3";
option go_package = "storj.io/storj/internalpb";
option go_package = "storj.io/storj/satellite/internalpb";
package metainfo;

41
satellite/internalpb/scannerValuer.go Normal file
View File

@ -0,0 +1,41 @@
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package internalpb
import (
"database/sql/driver"
proto "github.com/gogo/protobuf/proto"
"github.com/zeebo/errs"
"storj.io/common/pb"
)
var scanError = errs.Class("Protobuf Scanner")
var valueError = errs.Class("Protobuf Valuer")
// scan automatically converts database []byte to proto.Messages.
func scan(msg proto.Message, value interface{}) error {
bytes, ok := value.([]byte)
if !ok {
return scanError.New("%t was %t, expected []bytes", msg, value)
}
return scanError.Wrap(pb.Unmarshal(bytes, msg))
}
// value automatically converts proto.Messages to database []byte.
func value(msg proto.Message) (driver.Value, error) {
value, err := pb.Marshal(msg)
return value, valueError.Wrap(err)
}
// Scan implements the Scanner interface.
func (n *InjuredSegment) Scan(value interface{}) error {
return scan(n, value)
}
// Value implements the driver Valuer interface.
func (n InjuredSegment) Value() (driver.Value, error) {
return value(&n)
}

5
satellite/repair/checker/checker.go
View File

@ -15,6 +15,7 @@ import (
"storj.io/common/errs2"
"storj.io/common/pb"
"storj.io/common/sync2"
"storj.io/storj/satellite/internalpb"
"storj.io/storj/satellite/metainfo"
"storj.io/storj/satellite/metainfo/metabase"
"storj.io/storj/satellite/overlay"
@ -204,7 +205,7 @@ func (checker *Checker) updateIrreparableSegmentStatus(ctx context.Context, poin
// If the segment is suddenly entirely healthy again, we don't need to repair and we don't need to
// keep it in the irreparabledb queue either.
if numHealthy >= redundancy.MinReq && numHealthy <= repairThreshold && numHealthy < redundancy.SuccessThreshold {
_, err = checker.repairQueue.Insert(ctx, &pb.InjuredSegment{
_, err = checker.repairQueue.Insert(ctx, &internalpb.InjuredSegment{
Path: key,
LostPieces: missingPieces,
InsertedTime: time.Now().UTC(),
@ -308,7 +309,7 @@ func (obs *checkerObserver) RemoteSegment(ctx context.Context, segment *metainfo
// except for the case when the repair and success thresholds are the same (a case usually seen during testing)
if numHealthy >= required && numHealthy <= repairThreshold && numHealthy < successThreshold {
obs.monStats.remoteSegmentsNeedingRepair++
alreadyInserted, err := obs.repairQueue.Insert(ctx, &pb.InjuredSegment{
alreadyInserted, err := obs.repairQueue.Insert(ctx, &internalpb.InjuredSegment{
Path: key,
LostPieces: missingPieces,
InsertedTime: time.Now().UTC(),

10
satellite/repair/queue/queue.go
View File

@ -7,7 +7,7 @@ import (
"context"
"time"
"storj.io/common/pb"
"storj.io/storj/satellite/internalpb"
)
// RepairQueue implements queueing for segments that need repairing.
@ -16,15 +16,15 @@ import (
// architecture: Database
type RepairQueue interface {
// Insert adds an injured segment.
Insert(ctx context.Context, s *pb.InjuredSegment, numHealthy int) (alreadyInserted bool, err error)
Insert(ctx context.Context, s *internalpb.InjuredSegment, numHealthy int) (alreadyInserted bool, err error)
// Select gets an injured segment.
Select(ctx context.Context) (*pb.InjuredSegment, error)
Select(ctx context.Context) (*internalpb.InjuredSegment, error)
// Delete removes an injured segment.
Delete(ctx context.Context, s *pb.InjuredSegment) error
Delete(ctx context.Context, s *internalpb.InjuredSegment) error
// Clean removes all segments last updated before a certain time
Clean(ctx context.Context, before time.Time) (deleted int64, err error)
// SelectN lists limit amount of injured segments.
SelectN(ctx context.Context, limit int) ([]pb.InjuredSegment, error)
SelectN(ctx context.Context, limit int) ([]internalpb.InjuredSegment, error)
// Count counts the number of segments in the repair queue.
Count(ctx context.Context) (count int, err error)
}

12
satellite/repair/queue/queue2_test.go
View File

@ -13,9 +13,9 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"storj.io/common/pb"
"storj.io/common/testcontext"
"storj.io/storj/satellite"
"storj.io/storj/satellite/internalpb"
"storj.io/storj/satellite/satellitedb/dbx"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
"storj.io/storj/storage"
@ -29,7 +29,7 @@ func TestUntilEmpty(t *testing.T) {
pathsMap := make(map[string]int)
for i := 0; i < 20; i++ {
path := "/path/" + strconv.Itoa(i)
injuredSeg := &pb.InjuredSegment{Path: []byte(path)}
injuredSeg := &internalpb.InjuredSegment{Path: []byte(path)}
alreadyInserted, err := repairQueue.Insert(ctx, injuredSeg, 10)
require.NoError(t, err)
require.False(t, alreadyInserted)
@ -62,7 +62,7 @@ func TestOrder(t *testing.T) {
olderRepairPath := []byte("/path/older")
for _, path := range [][]byte{oldRepairPath, recentRepairPath, nullPath, olderRepairPath} {
injuredSeg := &pb.InjuredSegment{Path: path}
injuredSeg := &internalpb.InjuredSegment{Path: path}
alreadyInserted, err := repairQueue.Insert(ctx, injuredSeg, 10)
require.NoError(t, err)
require.False(t, alreadyInserted)
@ -157,7 +157,7 @@ func TestOrderHealthyPieces(t *testing.T) {
})
for _, item := range injuredSegList {
// first, insert the injured segment
injuredSeg := &pb.InjuredSegment{Path: item.path}
injuredSeg := &internalpb.InjuredSegment{Path: item.path}
alreadyInserted, err := repairQueue.Insert(ctx, injuredSeg, item.health)
require.NoError(t, err)
require.False(t, alreadyInserted)
@ -221,7 +221,7 @@ func TestOrderOverwrite(t *testing.T) {
{[]byte("path/a"), 8},
}
for i, item := range injuredSegList {
injuredSeg := &pb.InjuredSegment{Path: item.path}
injuredSeg := &internalpb.InjuredSegment{Path: item.path}
alreadyInserted, err := repairQueue.Insert(ctx, injuredSeg, item.health)
require.NoError(t, err)
if i == 2 {
@ -256,7 +256,7 @@ func TestCount(t *testing.T) {
numSegments := 20
for i := 0; i < numSegments; i++ {
path := "/path/" + strconv.Itoa(i)
injuredSeg := &pb.InjuredSegment{Path: []byte(path)}
injuredSeg := &internalpb.InjuredSegment{Path: []byte(path)}
alreadyInserted, err := repairQueue.Insert(ctx, injuredSeg, 10)
require.NoError(t, err)
require.False(t, alreadyInserted)

21
satellite/repair/queue/queue_test.go
View File

@ -15,6 +15,7 @@ import (
"storj.io/common/pb"
"storj.io/common/testcontext"
"storj.io/storj/satellite"
"storj.io/storj/satellite/internalpb"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
"storj.io/storj/storage"
)
@ -23,7 +24,7 @@ func TestInsertSelect(t *testing.T) {
satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
q := db.RepairQueue()
seg := &pb.InjuredSegment{
seg := &internalpb.InjuredSegment{
Path: []byte("abc"),
LostPieces: []int32{int32(1), int32(3)},
}
@ -42,7 +43,7 @@ func TestInsertDuplicate(t *testing.T) {
satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
q := db.RepairQueue()
seg := &pb.InjuredSegment{
seg := &internalpb.InjuredSegment{
Path: []byte("abc"),
LostPieces: []int32{int32(1), int32(3)},
}
@ -70,9 +71,9 @@ func TestSequential(t *testing.T) {
q := db.RepairQueue()
const N = 20
var addSegs []*pb.InjuredSegment
var addSegs []*internalpb.InjuredSegment
for i := 0; i < N; i++ {
seg := &pb.InjuredSegment{
seg := &internalpb.InjuredSegment{
Path: []byte(strconv.Itoa(i)),
LostPieces: []int32{int32(i)},
}
@ -108,14 +109,14 @@ func TestParallel(t *testing.T) {
satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
q := db.RepairQueue()
const N = 20
entries := make(chan *pb.InjuredSegment, N)
entries := make(chan *internalpb.InjuredSegment, N)
var inserts errs2.Group
// Add to queue concurrently
for i := 0; i < N; i++ {
i := i
inserts.Go(func() error {
_, err := q.Insert(ctx, &pb.InjuredSegment{
_, err := q.Insert(ctx, &internalpb.InjuredSegment{
Path: []byte(strconv.Itoa(i)),
LostPieces: []int32{int32(i)},
}, 10)
@ -146,7 +147,7 @@ func TestParallel(t *testing.T) {
require.Empty(t, remove.Wait(), "unexpected queue.Select/Delete errors")
close(entries)
var items []*pb.InjuredSegment
var items []*internalpb.InjuredSegment
for segment := range entries {
items = append(items, segment)
}
@ -166,15 +167,15 @@ func TestClean(t *testing.T) {
satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
q := db.RepairQueue()
seg1 := &pb.InjuredSegment{
seg1 := &internalpb.InjuredSegment{
Path: []byte("seg1"),
LostPieces: []int32{int32(1), int32(3)},
}
seg2 := &pb.InjuredSegment{
seg2 := &internalpb.InjuredSegment{
Path: []byte("seg2"),
LostPieces: []int32{int32(1), int32(3)},
}
seg3 := &pb.InjuredSegment{
seg3 := &internalpb.InjuredSegment{
Path: []byte("seg3"),
LostPieces: []int32{int32(1), int32(3)},
}

3
satellite/repair/repairer/repairer.go
View File

@ -15,6 +15,7 @@ import (
"storj.io/common/memory"
"storj.io/common/pb"
"storj.io/common/sync2"
"storj.io/storj/satellite/internalpb"
"storj.io/storj/satellite/repair/irreparable"
"storj.io/storj/satellite/repair/queue"
"storj.io/storj/storage"
@ -149,7 +150,7 @@ func (service *Service) process(ctx context.Context) (err error) {
return nil
}
func (service *Service) worker(ctx context.Context, seg *pb.InjuredSegment) (err error) {
func (service *Service) worker(ctx context.Context, seg *internalpb.InjuredSegment) (err error) {
defer mon.Task()(&ctx)(&err)
workerStartTime := time.Now().UTC()

12
satellite/satellitedb/repairqueue.go
View File

@ -11,8 +11,8 @@ import (
"github.com/zeebo/errs"
"storj.io/common/pb"
"storj.io/storj/private/dbutil"
"storj.io/storj/satellite/internalpb"
"storj.io/storj/satellite/satellitedb/dbx"
"storj.io/storj/storage"
)
@ -24,7 +24,7 @@ type repairQueue struct {
db *satelliteDB
}
func (r *repairQueue) Insert(ctx context.Context, seg *pb.InjuredSegment, numHealthy int) (alreadyInserted bool, err error) {
func (r *repairQueue) Insert(ctx context.Context, seg *internalpb.InjuredSegment, numHealthy int) (alreadyInserted bool, err error) {
defer mon.Task()(&ctx)(&err)
// insert if not exists, or update healthy count if does exist
var query string
@ -77,7 +77,7 @@ func (r *repairQueue) Insert(ctx context.Context, seg *pb.InjuredSegment, numHea
return alreadyInserted, rows.Err()
}
func (r *repairQueue) Select(ctx context.Context) (seg *pb.InjuredSegment, err error) {
func (r *repairQueue) Select(ctx context.Context) (seg *internalpb.InjuredSegment, err error) {
defer mon.Task()(&ctx)(&err)
switch r.db.implementation {
case dbutil.Cockroach:
@ -103,7 +103,7 @@ func (r *repairQueue) Select(ctx context.Context) (seg *pb.InjuredSegment, err e
return seg, err
}
func (r *repairQueue) Delete(ctx context.Context, seg *pb.InjuredSegment) (err error) {
func (r *repairQueue) Delete(ctx context.Context, seg *internalpb.InjuredSegment) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM injuredsegments WHERE path = ?`), seg.Path)
return Error.Wrap(err)
@ -115,7 +115,7 @@ func (r *repairQueue) Clean(ctx context.Context, before time.Time) (deleted int6
return n, Error.Wrap(err)
}
func (r *repairQueue) SelectN(ctx context.Context, limit int) (segs []pb.InjuredSegment, err error) {
func (r *repairQueue) SelectN(ctx context.Context, limit int) (segs []internalpb.InjuredSegment, err error) {
defer mon.Task()(&ctx)(&err)
if limit <= 0 || limit > RepairQueueSelectLimit {
limit = RepairQueueSelectLimit
@ -128,7 +128,7 @@ func (r *repairQueue) SelectN(ctx context.Context, limit int) (segs []pb.Injured
defer func() { err = errs.Combine(err, rows.Close()) }()
for rows.Next() {
var seg pb.InjuredSegment
var seg internalpb.InjuredSegment
err = rows.Scan(&seg)
if err != nil {
return segs, Error.Wrap(err)