mirror of
				https://github.com/minio/minio.git
				synced 2025-10-29 15:55:00 -04:00 
			
		
		
		
	fix: allow DNS disconnection events to happen in k8s (#19145)
in k8s things really do come online very asynchronously, we need to use implementation that allows this randomness. To facilitate this move WriteAll() as part of the websocket layer instead. Bonus: avoid instances of dnscache usage on k8s
This commit is contained in:
		
							parent
							
								
									62ce52c8fd
								
							
						
					
					
						commit
						51874a5776
					
				| @ -751,7 +751,12 @@ func serverHandleEnvVars() { | ||||
| 		for _, endpoint := range minioEndpoints { | ||||
| 			if net.ParseIP(endpoint) == nil { | ||||
| 				// Checking if the IP is a DNS entry. | ||||
| 				addrs, err := globalDNSCache.LookupHost(GlobalContext, endpoint) | ||||
| 				lookupHost := globalDNSCache.LookupHost | ||||
| 				if IsKubernetes() || IsDocker() { | ||||
| 					lookupHost = net.DefaultResolver.LookupHost | ||||
| 				} | ||||
| 
 | ||||
| 				addrs, err := lookupHost(GlobalContext, endpoint) | ||||
| 				if err != nil { | ||||
| 					logger.FatalIf(err, "Unable to initialize MinIO server with [%s] invalid entry found in MINIO_PUBLIC_IPS", endpoint) | ||||
| 				} | ||||
|  | ||||
| @ -98,7 +98,12 @@ func mustGetLocalIP6() (ipList set.StringSet) { | ||||
| 
 | ||||
| // getHostIP returns IP address of given host. | ||||
| func getHostIP(host string) (ipList set.StringSet, err error) { | ||||
| 	addrs, err := globalDNSCache.LookupHost(GlobalContext, host) | ||||
| 	lookupHost := globalDNSCache.LookupHost | ||||
| 	if IsKubernetes() || IsDocker() { | ||||
| 		lookupHost = net.DefaultResolver.LookupHost | ||||
| 	} | ||||
| 
 | ||||
| 	addrs, err := lookupHost(GlobalContext, host) | ||||
| 	if err != nil { | ||||
| 		return ipList, err | ||||
| 	} | ||||
|  | ||||
| @ -581,7 +581,12 @@ func setGlobalInternodeInterface(interfaceName string) { | ||||
| 				ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||||
| 				defer cancel() | ||||
| 
 | ||||
| 				haddrs, err := globalDNSCache.LookupHost(ctx, host) | ||||
| 				lookupHost := globalDNSCache.LookupHost | ||||
| 				if IsKubernetes() || IsDocker() { | ||||
| 					lookupHost = net.DefaultResolver.LookupHost | ||||
| 				} | ||||
| 
 | ||||
| 				haddrs, err := lookupHost(ctx, host) | ||||
| 				if err == nil { | ||||
| 					ip = haddrs[0] | ||||
| 				} | ||||
| @ -619,7 +624,12 @@ func getServerListenAddrs() []string { | ||||
| 		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||||
| 		defer cancel() | ||||
| 
 | ||||
| 		haddrs, err := globalDNSCache.LookupHost(ctx, host) | ||||
| 		lookupHost := globalDNSCache.LookupHost | ||||
| 		if IsKubernetes() || IsDocker() { | ||||
| 			lookupHost = net.DefaultResolver.LookupHost | ||||
| 		} | ||||
| 
 | ||||
| 		haddrs, err := lookupHost(ctx, host) | ||||
| 		if err == nil { | ||||
| 			for _, addr := range haddrs { | ||||
| 				addrs.Add(net.JoinHostPort(addr, globalMinioPort)) | ||||
|  | ||||
| @ -436,6 +436,14 @@ type ReadAllHandlerParams struct { | ||||
| 	FilePath string `msg:"fp"` | ||||
| } | ||||
| 
 | ||||
| // WriteAllHandlerParams are parameters for WriteAllHandler. | ||||
| type WriteAllHandlerParams struct { | ||||
| 	DiskID   string `msg:"id"` | ||||
| 	Volume   string `msg:"v"` | ||||
| 	FilePath string `msg:"fp"` | ||||
| 	Buf      []byte `msg:"b"` | ||||
| } | ||||
| 
 | ||||
| // RenameDataResp - RenameData()'s response. | ||||
| type RenameDataResp struct { | ||||
| 	Signature uint64 `msg:"sig"` | ||||
|  | ||||
| @ -5375,3 +5375,181 @@ func (z VolsInfo) Msgsize() (s int) { | ||||
| 	} | ||||
| 	return | ||||
| } | ||||
| 
 | ||||
| // DecodeMsg implements msgp.Decodable | ||||
| func (z *WriteAllHandlerParams) DecodeMsg(dc *msgp.Reader) (err error) { | ||||
| 	var field []byte | ||||
| 	_ = field | ||||
| 	var zb0001 uint32 | ||||
| 	zb0001, err = dc.ReadMapHeader() | ||||
| 	if err != nil { | ||||
| 		err = msgp.WrapError(err) | ||||
| 		return | ||||
| 	} | ||||
| 	for zb0001 > 0 { | ||||
| 		zb0001-- | ||||
| 		field, err = dc.ReadMapKeyPtr() | ||||
| 		if err != nil { | ||||
| 			err = msgp.WrapError(err) | ||||
| 			return | ||||
| 		} | ||||
| 		switch msgp.UnsafeString(field) { | ||||
| 		case "id": | ||||
| 			z.DiskID, err = dc.ReadString() | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "DiskID") | ||||
| 				return | ||||
| 			} | ||||
| 		case "v": | ||||
| 			z.Volume, err = dc.ReadString() | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "Volume") | ||||
| 				return | ||||
| 			} | ||||
| 		case "fp": | ||||
| 			z.FilePath, err = dc.ReadString() | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "FilePath") | ||||
| 				return | ||||
| 			} | ||||
| 		case "b": | ||||
| 			z.Buf, err = dc.ReadBytes(z.Buf) | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "Buf") | ||||
| 				return | ||||
| 			} | ||||
| 		default: | ||||
| 			err = dc.Skip() | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err) | ||||
| 				return | ||||
| 			} | ||||
| 		} | ||||
| 	} | ||||
| 	return | ||||
| } | ||||
| 
 | ||||
| // EncodeMsg implements msgp.Encodable | ||||
| func (z *WriteAllHandlerParams) EncodeMsg(en *msgp.Writer) (err error) { | ||||
| 	// map header, size 4 | ||||
| 	// write "id" | ||||
| 	err = en.Append(0x84, 0xa2, 0x69, 0x64) | ||||
| 	if err != nil { | ||||
| 		return | ||||
| 	} | ||||
| 	err = en.WriteString(z.DiskID) | ||||
| 	if err != nil { | ||||
| 		err = msgp.WrapError(err, "DiskID") | ||||
| 		return | ||||
| 	} | ||||
| 	// write "v" | ||||
| 	err = en.Append(0xa1, 0x76) | ||||
| 	if err != nil { | ||||
| 		return | ||||
| 	} | ||||
| 	err = en.WriteString(z.Volume) | ||||
| 	if err != nil { | ||||
| 		err = msgp.WrapError(err, "Volume") | ||||
| 		return | ||||
| 	} | ||||
| 	// write "fp" | ||||
| 	err = en.Append(0xa2, 0x66, 0x70) | ||||
| 	if err != nil { | ||||
| 		return | ||||
| 	} | ||||
| 	err = en.WriteString(z.FilePath) | ||||
| 	if err != nil { | ||||
| 		err = msgp.WrapError(err, "FilePath") | ||||
| 		return | ||||
| 	} | ||||
| 	// write "b" | ||||
| 	err = en.Append(0xa1, 0x62) | ||||
| 	if err != nil { | ||||
| 		return | ||||
| 	} | ||||
| 	err = en.WriteBytes(z.Buf) | ||||
| 	if err != nil { | ||||
| 		err = msgp.WrapError(err, "Buf") | ||||
| 		return | ||||
| 	} | ||||
| 	return | ||||
| } | ||||
| 
 | ||||
| // MarshalMsg implements msgp.Marshaler | ||||
| func (z *WriteAllHandlerParams) MarshalMsg(b []byte) (o []byte, err error) { | ||||
| 	o = msgp.Require(b, z.Msgsize()) | ||||
| 	// map header, size 4 | ||||
| 	// string "id" | ||||
| 	o = append(o, 0x84, 0xa2, 0x69, 0x64) | ||||
| 	o = msgp.AppendString(o, z.DiskID) | ||||
| 	// string "v" | ||||
| 	o = append(o, 0xa1, 0x76) | ||||
| 	o = msgp.AppendString(o, z.Volume) | ||||
| 	// string "fp" | ||||
| 	o = append(o, 0xa2, 0x66, 0x70) | ||||
| 	o = msgp.AppendString(o, z.FilePath) | ||||
| 	// string "b" | ||||
| 	o = append(o, 0xa1, 0x62) | ||||
| 	o = msgp.AppendBytes(o, z.Buf) | ||||
| 	return | ||||
| } | ||||
| 
 | ||||
| // UnmarshalMsg implements msgp.Unmarshaler | ||||
| func (z *WriteAllHandlerParams) UnmarshalMsg(bts []byte) (o []byte, err error) { | ||||
| 	var field []byte | ||||
| 	_ = field | ||||
| 	var zb0001 uint32 | ||||
| 	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts) | ||||
| 	if err != nil { | ||||
| 		err = msgp.WrapError(err) | ||||
| 		return | ||||
| 	} | ||||
| 	for zb0001 > 0 { | ||||
| 		zb0001-- | ||||
| 		field, bts, err = msgp.ReadMapKeyZC(bts) | ||||
| 		if err != nil { | ||||
| 			err = msgp.WrapError(err) | ||||
| 			return | ||||
| 		} | ||||
| 		switch msgp.UnsafeString(field) { | ||||
| 		case "id": | ||||
| 			z.DiskID, bts, err = msgp.ReadStringBytes(bts) | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "DiskID") | ||||
| 				return | ||||
| 			} | ||||
| 		case "v": | ||||
| 			z.Volume, bts, err = msgp.ReadStringBytes(bts) | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "Volume") | ||||
| 				return | ||||
| 			} | ||||
| 		case "fp": | ||||
| 			z.FilePath, bts, err = msgp.ReadStringBytes(bts) | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "FilePath") | ||||
| 				return | ||||
| 			} | ||||
| 		case "b": | ||||
| 			z.Buf, bts, err = msgp.ReadBytesBytes(bts, z.Buf) | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err, "Buf") | ||||
| 				return | ||||
| 			} | ||||
| 		default: | ||||
| 			bts, err = msgp.Skip(bts) | ||||
| 			if err != nil { | ||||
| 				err = msgp.WrapError(err) | ||||
| 				return | ||||
| 			} | ||||
| 		} | ||||
| 	} | ||||
| 	o = bts | ||||
| 	return | ||||
| } | ||||
| 
 | ||||
| // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message | ||||
| func (z *WriteAllHandlerParams) Msgsize() (s int) { | ||||
| 	s = 1 + 3 + msgp.StringPrefixSize + len(z.DiskID) + 2 + msgp.StringPrefixSize + len(z.Volume) + 3 + msgp.StringPrefixSize + len(z.FilePath) + 2 + msgp.BytesPrefixSize + len(z.Buf) | ||||
| 	return | ||||
| } | ||||
|  | ||||
| @ -2833,3 +2833,116 @@ func BenchmarkDecodeVolsInfo(b *testing.B) { | ||||
| 		} | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| func TestMarshalUnmarshalWriteAllHandlerParams(t *testing.T) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	bts, err := v.MarshalMsg(nil) | ||||
| 	if err != nil { | ||||
| 		t.Fatal(err) | ||||
| 	} | ||||
| 	left, err := v.UnmarshalMsg(bts) | ||||
| 	if err != nil { | ||||
| 		t.Fatal(err) | ||||
| 	} | ||||
| 	if len(left) > 0 { | ||||
| 		t.Errorf("%d bytes left over after UnmarshalMsg(): %q", len(left), left) | ||||
| 	} | ||||
| 
 | ||||
| 	left, err = msgp.Skip(bts) | ||||
| 	if err != nil { | ||||
| 		t.Fatal(err) | ||||
| 	} | ||||
| 	if len(left) > 0 { | ||||
| 		t.Errorf("%d bytes left over after Skip(): %q", len(left), left) | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| func BenchmarkMarshalMsgWriteAllHandlerParams(b *testing.B) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	b.ReportAllocs() | ||||
| 	b.ResetTimer() | ||||
| 	for i := 0; i < b.N; i++ { | ||||
| 		v.MarshalMsg(nil) | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| func BenchmarkAppendMsgWriteAllHandlerParams(b *testing.B) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	bts := make([]byte, 0, v.Msgsize()) | ||||
| 	bts, _ = v.MarshalMsg(bts[0:0]) | ||||
| 	b.SetBytes(int64(len(bts))) | ||||
| 	b.ReportAllocs() | ||||
| 	b.ResetTimer() | ||||
| 	for i := 0; i < b.N; i++ { | ||||
| 		bts, _ = v.MarshalMsg(bts[0:0]) | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| func BenchmarkUnmarshalWriteAllHandlerParams(b *testing.B) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	bts, _ := v.MarshalMsg(nil) | ||||
| 	b.ReportAllocs() | ||||
| 	b.SetBytes(int64(len(bts))) | ||||
| 	b.ResetTimer() | ||||
| 	for i := 0; i < b.N; i++ { | ||||
| 		_, err := v.UnmarshalMsg(bts) | ||||
| 		if err != nil { | ||||
| 			b.Fatal(err) | ||||
| 		} | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| func TestEncodeDecodeWriteAllHandlerParams(t *testing.T) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	var buf bytes.Buffer | ||||
| 	msgp.Encode(&buf, &v) | ||||
| 
 | ||||
| 	m := v.Msgsize() | ||||
| 	if buf.Len() > m { | ||||
| 		t.Log("WARNING: TestEncodeDecodeWriteAllHandlerParams Msgsize() is inaccurate") | ||||
| 	} | ||||
| 
 | ||||
| 	vn := WriteAllHandlerParams{} | ||||
| 	err := msgp.Decode(&buf, &vn) | ||||
| 	if err != nil { | ||||
| 		t.Error(err) | ||||
| 	} | ||||
| 
 | ||||
| 	buf.Reset() | ||||
| 	msgp.Encode(&buf, &v) | ||||
| 	err = msgp.NewReader(&buf).Skip() | ||||
| 	if err != nil { | ||||
| 		t.Error(err) | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| func BenchmarkEncodeWriteAllHandlerParams(b *testing.B) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	var buf bytes.Buffer | ||||
| 	msgp.Encode(&buf, &v) | ||||
| 	b.SetBytes(int64(buf.Len())) | ||||
| 	en := msgp.NewWriter(msgp.Nowhere) | ||||
| 	b.ReportAllocs() | ||||
| 	b.ResetTimer() | ||||
| 	for i := 0; i < b.N; i++ { | ||||
| 		v.EncodeMsg(en) | ||||
| 	} | ||||
| 	en.Flush() | ||||
| } | ||||
| 
 | ||||
| func BenchmarkDecodeWriteAllHandlerParams(b *testing.B) { | ||||
| 	v := WriteAllHandlerParams{} | ||||
| 	var buf bytes.Buffer | ||||
| 	msgp.Encode(&buf, &v) | ||||
| 	b.SetBytes(int64(buf.Len())) | ||||
| 	rd := msgp.NewEndlessReader(buf.Bytes(), b) | ||||
| 	dc := msgp.NewReader(rd) | ||||
| 	b.ReportAllocs() | ||||
| 	b.ResetTimer() | ||||
| 	for i := 0; i < b.N; i++ { | ||||
| 		err := v.DecodeMsg(dc) | ||||
| 		if err != nil { | ||||
| 			b.Fatal(err) | ||||
| 		} | ||||
| 	} | ||||
| } | ||||
|  | ||||
| @ -205,7 +205,7 @@ func (client *storageRESTClient) String() string { | ||||
| 
 | ||||
| // IsOnline - returns whether RPC client failed to connect or not. | ||||
| func (client *storageRESTClient) IsOnline() bool { | ||||
| 	return client.restClient.IsOnline() && client.gridConn.State() == grid.StateConnected | ||||
| 	return client.gridConn.State() == grid.StateConnected | ||||
| } | ||||
| 
 | ||||
| // LastConn - returns when the disk is seen to be connected the last time | ||||
| @ -295,7 +295,7 @@ func (client *storageRESTClient) SetDiskID(id string) { | ||||
| } | ||||
| 
 | ||||
| func (client *storageRESTClient) DiskInfo(ctx context.Context, opts DiskInfoOptions) (info DiskInfo, err error) { | ||||
| 	if client.gridConn.State() != grid.StateConnected { | ||||
| 	if !client.IsOnline() { | ||||
| 		// make sure to check if the disk is offline, since the underlying | ||||
| 		// value is cached we should attempt to invalidate it if such calls | ||||
| 		// were attempted. This can lead to false success under certain conditions | ||||
| @ -442,12 +442,19 @@ func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path | ||||
| 
 | ||||
| // WriteAll - write all data to a file. | ||||
| func (client *storageRESTClient) WriteAll(ctx context.Context, volume string, path string, b []byte) error { | ||||
| 	values := make(url.Values) | ||||
| 	values.Set(storageRESTVolume, volume) | ||||
| 	values.Set(storageRESTFilePath, path) | ||||
| 	respBody, err := client.call(ctx, storageRESTMethodWriteAll, values, bytes.NewBuffer(b), int64(len(b))) | ||||
| 	defer xhttp.DrainBody(respBody) | ||||
| 	return err | ||||
| 	// Specific optimization to avoid re-read from the drives for `format.json` | ||||
| 	// in-case the caller is a network operation. | ||||
| 	if volume == minioMetaBucket && path == formatConfigFile { | ||||
| 		client.SetFormatData(b) | ||||
| 	} | ||||
| 
 | ||||
| 	_, err := storageWriteAllRPC.Call(ctx, client.gridConn, &WriteAllHandlerParams{ | ||||
| 		DiskID:   client.diskID, | ||||
| 		Volume:   volume, | ||||
| 		FilePath: path, | ||||
| 		Buf:      b, | ||||
| 	}) | ||||
| 	return toStorageErr(err) | ||||
| } | ||||
| 
 | ||||
| // CheckParts - stat all file parts. | ||||
| @ -883,7 +890,8 @@ func newStorageRESTClient(endpoint Endpoint, healthCheck bool, gm *grid.Manager) | ||||
| 		return nil, fmt.Errorf("unable to find connection for %s in targets: %v", endpoint.GridHost(), gm.Targets()) | ||||
| 	} | ||||
| 	return &storageRESTClient{ | ||||
| 		endpoint: endpoint, restClient: restClient, poolIndex: -1, setIndex: -1, diskIndex: -1, | ||||
| 		endpoint:      endpoint, | ||||
| 		restClient:    restClient, | ||||
| 		gridConn:      conn, | ||||
| 		diskInfoCache: cachevalue.New[DiskInfo](), | ||||
| 	}, nil | ||||
|  | ||||
| @ -64,6 +64,7 @@ var ( | ||||
| 	storageDiskInfoRPC       = grid.NewSingleHandler[*DiskInfoOptions, *DiskInfo](grid.HandlerDiskInfo, func() *DiskInfoOptions { return &DiskInfoOptions{} }, func() *DiskInfo { return &DiskInfo{} }).WithSharedResponse().AllowCallRequestPool(true) | ||||
| 	storageNSScannerRPC      = grid.NewStream[*nsScannerOptions, grid.NoPayload, *nsScannerResp](grid.HandlerNSScanner, func() *nsScannerOptions { return &nsScannerOptions{} }, nil, func() *nsScannerResp { return &nsScannerResp{} }) | ||||
| 	storageReadAllRPC        = grid.NewSingleHandler[*ReadAllHandlerParams, *grid.Bytes](grid.HandlerReadAll, func() *ReadAllHandlerParams { return &ReadAllHandlerParams{} }, grid.NewBytes).AllowCallRequestPool(true) | ||||
| 	storageWriteAllRPC       = grid.NewSingleHandler[*WriteAllHandlerParams, grid.NoPayload](grid.HandlerWriteAll, func() *WriteAllHandlerParams { return &WriteAllHandlerParams{} }, grid.NewNoPayload) | ||||
| 	storageReadVersionRPC    = grid.NewSingleHandler[*grid.MSS, *FileInfo](grid.HandlerReadVersion, grid.NewMSS, func() *FileInfo { return &FileInfo{} }) | ||||
| 	storageReadXLRPC         = grid.NewSingleHandler[*grid.MSS, *RawFileInfo](grid.HandlerReadXL, grid.NewMSS, func() *RawFileInfo { return &RawFileInfo{} }) | ||||
| 	storageRenameDataRPC     = grid.NewSingleHandler[*RenameDataHandlerParams, *RenameDataResp](grid.HandlerRenameData, func() *RenameDataHandlerParams { return &RenameDataHandlerParams{} }, func() *RenameDataResp { return &RenameDataResp{} }) | ||||
| @ -433,30 +434,6 @@ func (s *storageRESTServer) UpdateMetadataHandler(p *MetadataHandlerParams) (gri | ||||
| 	return grid.NewNPErr(s.getStorage().UpdateMetadata(context.Background(), volume, filePath, p.FI, p.UpdateOpts)) | ||||
| } | ||||
| 
 | ||||
| // WriteAllHandler - write to file all content. | ||||
| func (s *storageRESTServer) WriteAllHandler(w http.ResponseWriter, r *http.Request) { | ||||
| 	if !s.IsValid(w, r) { | ||||
| 		return | ||||
| 	} | ||||
| 	volume := r.Form.Get(storageRESTVolume) | ||||
| 	filePath := r.Form.Get(storageRESTFilePath) | ||||
| 
 | ||||
| 	if r.ContentLength < 0 { | ||||
| 		s.writeErrorResponse(w, errInvalidArgument) | ||||
| 		return | ||||
| 	} | ||||
| 	tmp := make([]byte, r.ContentLength) | ||||
| 	_, err := io.ReadFull(r.Body, tmp) | ||||
| 	if err != nil { | ||||
| 		s.writeErrorResponse(w, err) | ||||
| 		return | ||||
| 	} | ||||
| 	err = s.getStorage().WriteAll(r.Context(), volume, filePath, tmp) | ||||
| 	if err != nil { | ||||
| 		s.writeErrorResponse(w, err) | ||||
| 	} | ||||
| } | ||||
| 
 | ||||
| // CheckPartsHandler - check if a file metadata exists. | ||||
| func (s *storageRESTServer) CheckPartsHandler(p *CheckPartsHandlerParams) (grid.NoPayload, *grid.RemoteErr) { | ||||
| 	if !s.checkID(p.DiskID) { | ||||
| @ -467,6 +444,17 @@ func (s *storageRESTServer) CheckPartsHandler(p *CheckPartsHandlerParams) (grid. | ||||
| 	return grid.NewNPErr(s.getStorage().CheckParts(context.Background(), volume, filePath, p.FI)) | ||||
| } | ||||
| 
 | ||||
| func (s *storageRESTServer) WriteAllHandler(p *WriteAllHandlerParams) (grid.NoPayload, *grid.RemoteErr) { | ||||
| 	if !s.checkID(p.DiskID) { | ||||
| 		return grid.NewNPErr(errDiskNotFound) | ||||
| 	} | ||||
| 
 | ||||
| 	volume := p.Volume | ||||
| 	filePath := p.FilePath | ||||
| 
 | ||||
| 	return grid.NewNPErr(s.getStorage().WriteAll(context.Background(), volume, filePath, p.Buf)) | ||||
| } | ||||
| 
 | ||||
| // ReadAllHandler - read all the contents of a file. | ||||
| func (s *storageRESTServer) ReadAllHandler(p *ReadAllHandlerParams) (*grid.Bytes, *grid.RemoteErr) { | ||||
| 	if !s.checkID(p.DiskID) { | ||||
| @ -1298,8 +1286,6 @@ func registerStorageRESTHandlers(router *mux.Router, endpointServerPools Endpoin | ||||
| 
 | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodHealth).HandlerFunc(h(server.HealthHandler)) | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodAppendFile).HandlerFunc(h(server.AppendFileHandler)) | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodWriteAll).HandlerFunc(h(server.WriteAllHandler)) | ||||
| 
 | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodReadVersion).HandlerFunc(h(server.ReadVersionHandler)) | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodReadXL).HandlerFunc(h(server.ReadXLHandler)) | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodCreateFile).HandlerFunc(h(server.CreateFileHandler)) | ||||
| @ -1313,6 +1299,7 @@ func registerStorageRESTHandlers(router *mux.Router, endpointServerPools Endpoin | ||||
| 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodCleanAbandoned).HandlerFunc(h(server.CleanAbandonedDataHandler)) | ||||
| 			logger.FatalIf(storageListDirRPC.RegisterNoInput(gm, server.ListDirHandler, endpoint.Path), "unable to register handler") | ||||
| 			logger.FatalIf(storageReadAllRPC.Register(gm, server.ReadAllHandler, endpoint.Path), "unable to register handler") | ||||
| 			logger.FatalIf(storageWriteAllRPC.Register(gm, server.WriteAllHandler, endpoint.Path), "unable to register handler") | ||||
| 			logger.FatalIf(storageRenameFileRPC.Register(gm, server.RenameFileHandler, endpoint.Path), "unable to register handler") | ||||
| 			logger.FatalIf(storageRenameDataRPC.Register(gm, server.RenameDataHandler, endpoint.Path), "unable to register handler") | ||||
| 			logger.FatalIf(storageDeleteFileRPC.Register(gm, server.DeleteFileHandler, endpoint.Path), "unable to register handler") | ||||
|  | ||||
| @ -174,7 +174,7 @@ func (c ContextDialer) DialContext(ctx context.Context, network, address string) | ||||
| } | ||||
| 
 | ||||
| const ( | ||||
| 	defaultOutQueue    = 10000 | ||||
| 	defaultOutQueue    = 65535    // kind of close to max open fds per user | ||||
| 	readBufferSize     = 32 << 10 // 32 KiB is the most optimal on Linux | ||||
| 	writeBufferSize    = 32 << 10 // 32 KiB is the most optimal on Linux | ||||
| 	defaultDialTimeout = 2 * time.Second | ||||
|  | ||||
| @ -109,6 +109,7 @@ const ( | ||||
| 	HandlerGetLastDayTierStats | ||||
| 	HandlerSignalService | ||||
| 	HandlerGetBandwidth | ||||
| 	HandlerWriteAll | ||||
| 
 | ||||
| 	// Add more above here ^^^ | ||||
| 	// If all handlers are used, the type of Handler can be changed. | ||||
| @ -141,6 +142,7 @@ var handlerPrefixes = [handlerLast]string{ | ||||
| 	HandlerRenameData:                  storagePrefix, | ||||
| 	HandlerRenameFile:                  storagePrefix, | ||||
| 	HandlerReadAll:                     storagePrefix, | ||||
| 	HandlerWriteAll:                    storagePrefix, | ||||
| 	HandlerServerVerify:                bootstrapPrefix, | ||||
| 	HandlerTrace:                       peerPrefix, | ||||
| 	HandlerListen:                      peerPrefix, | ||||
|  | ||||
| @ -78,14 +78,15 @@ func _() { | ||||
| 	_ = x[HandlerGetLastDayTierStats-67] | ||||
| 	_ = x[HandlerSignalService-68] | ||||
| 	_ = x[HandlerGetBandwidth-69] | ||||
| 	_ = x[handlerTest-70] | ||||
| 	_ = x[handlerTest2-71] | ||||
| 	_ = x[handlerLast-72] | ||||
| 	_ = x[HandlerWriteAll-70] | ||||
| 	_ = x[handlerTest-71] | ||||
| 	_ = x[handlerTest2-72] | ||||
| 	_ = x[handlerLast-73] | ||||
| } | ||||
| 
 | ||||
| const _HandlerID_name = "handlerInvalidLockLockLockRLockLockUnlockLockRUnlockLockRefreshLockForceUnlockWalkDirStatVolDiskInfoNSScannerReadXLReadVersionDeleteFileDeleteVersionUpdateMetadataWriteMetadataCheckPartsRenameDataRenameFileReadAllServerVerifyTraceListenDeleteBucketMetadataLoadBucketMetadataReloadSiteReplicationConfigReloadPoolMetaStopRebalanceLoadRebalanceMetaLoadTransitionTierConfigDeletePolicyLoadPolicyLoadPolicyMappingDeleteServiceAccountLoadServiceAccountDeleteUserLoadUserLoadGroupHealBucketMakeBucketHeadBucketDeleteBucketGetMetricsGetResourceMetricsGetMemInfoGetProcInfoGetOSInfoGetPartitionsGetNetInfoGetCPUsServerInfoGetSysConfigGetSysServicesGetSysErrorsGetAllBucketStatsGetBucketStatsGetSRMetricsGetPeerMetricsGetMetacacheListingUpdateMetacacheListingGetPeerBucketMetricsStorageInfoConsoleLogListDirGetLocksBackgroundHealStatusGetLastDayTierStatsSignalServiceGetBandwidthhandlerTesthandlerTest2handlerLast" | ||||
| const _HandlerID_name = "handlerInvalidLockLockLockRLockLockUnlockLockRUnlockLockRefreshLockForceUnlockWalkDirStatVolDiskInfoNSScannerReadXLReadVersionDeleteFileDeleteVersionUpdateMetadataWriteMetadataCheckPartsRenameDataRenameFileReadAllServerVerifyTraceListenDeleteBucketMetadataLoadBucketMetadataReloadSiteReplicationConfigReloadPoolMetaStopRebalanceLoadRebalanceMetaLoadTransitionTierConfigDeletePolicyLoadPolicyLoadPolicyMappingDeleteServiceAccountLoadServiceAccountDeleteUserLoadUserLoadGroupHealBucketMakeBucketHeadBucketDeleteBucketGetMetricsGetResourceMetricsGetMemInfoGetProcInfoGetOSInfoGetPartitionsGetNetInfoGetCPUsServerInfoGetSysConfigGetSysServicesGetSysErrorsGetAllBucketStatsGetBucketStatsGetSRMetricsGetPeerMetricsGetMetacacheListingUpdateMetacacheListingGetPeerBucketMetricsStorageInfoConsoleLogListDirGetLocksBackgroundHealStatusGetLastDayTierStatsSignalServiceGetBandwidthWriteAllhandlerTesthandlerTest2handlerLast" | ||||
| 
 | ||||
| var _HandlerID_index = [...]uint16{0, 14, 22, 31, 41, 52, 63, 78, 85, 92, 100, 109, 115, 126, 136, 149, 163, 176, 186, 196, 206, 213, 225, 230, 236, 256, 274, 301, 315, 328, 345, 369, 381, 391, 408, 428, 446, 456, 464, 473, 483, 493, 503, 515, 525, 543, 553, 564, 573, 586, 596, 603, 613, 625, 639, 651, 668, 682, 694, 708, 727, 749, 769, 780, 790, 797, 805, 825, 844, 857, 869, 880, 892, 903} | ||||
| var _HandlerID_index = [...]uint16{0, 14, 22, 31, 41, 52, 63, 78, 85, 92, 100, 109, 115, 126, 136, 149, 163, 176, 186, 196, 206, 213, 225, 230, 236, 256, 274, 301, 315, 328, 345, 369, 381, 391, 408, 428, 446, 456, 464, 473, 483, 493, 503, 515, 525, 543, 553, 564, 573, 586, 596, 603, 613, 625, 639, 651, 668, 682, 694, 708, 727, 749, 769, 780, 790, 797, 805, 825, 844, 857, 869, 877, 888, 900, 911} | ||||
| 
 | ||||
| func (i HandlerID) String() string { | ||||
| 	if i >= HandlerID(len(_HandlerID_index)-1) { | ||||
|  | ||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user