// Create locators for testing.
// Turn on permission settings so we can generate signed locators.
- enforce_permissions = true
+ enforcePermissions = true
PermissionSecret = []byte(known_key)
blob_signature_ttl = 300 * time.Second
// -----------------
// Test unauthenticated request with permissions off.
- enforce_permissions = false
+ enforcePermissions = false
// Unauthenticated request, unsigned locator
// => OK
// ----------------
// Permissions: on.
- enforce_permissions = true
+ enforcePermissions = true
// Authenticated request, signed locator
// => OK
// - authenticated /index/prefix request | superuser
//
// The only /index requests that should succeed are those issued by the
-// superuser. They should pass regardless of the value of enforce_permissions.
+// superuser. They should pass regardless of the value of enforcePermissions.
//
func TestIndexHandler(t *testing.T) {
defer teardown()
// Only the superuser should be allowed to issue /index requests.
// ---------------------------
- // enforce_permissions enabled
+ // enforcePermissions enabled
// This setting should not affect tests passing.
- enforce_permissions = true
+ enforcePermissions = true
// unauthenticated /index request
// => UnauthorizedError
response := IssueRequest(unauthenticatedReq)
ExpectStatusCode(t,
- "enforce_permissions on, unauthenticated request",
+ "enforcePermissions on, unauthenticated request",
UnauthorizedError.HTTPCode,
response)
response)
// ----------------------------
- // enforce_permissions disabled
+ // enforcePermissions disabled
// Valid Request should still pass.
- enforce_permissions = false
+ enforcePermissions = false
// superuser /index request
// => OK
defer func(orig *bufferPool) {
bufs = orig
}(bufs)
- bufs = newBufferPool(1, BLOCKSIZE)
+ bufs = newBufferPool(1, BlockSize)
ok := make(chan struct{})
go func() {
return rest
}
+// BadRequestHandler is a HandleFunc to address bad requests.
func BadRequestHandler(w http.ResponseWriter, r *http.Request) {
http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode)
}
+// GetBlockHandler is a HandleFunc to address Get block requests.
func GetBlockHandler(resp http.ResponseWriter, req *http.Request) {
- if enforce_permissions {
+ if enforcePermissions {
locator := req.URL.Path[1:] // strip leading slash
if err := VerifySignature(locator, GetApiToken(req)); err != nil {
http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
resp.Write(block)
}
+// PutBlockHandler is a HandleFunc to address Put block requests.
func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
hash := mux.Vars(req)["hash"]
return
}
- if req.ContentLength > BLOCKSIZE {
+ if req.ContentLength > BlockSize {
http.Error(resp, TooLongError.Error(), TooLongError.HTTPCode)
return
}
// Success; add a size hint, sign the locator if possible, and
// return it to the client.
- return_hash := fmt.Sprintf("%s+%d", hash, req.ContentLength)
- api_token := GetApiToken(req)
- if PermissionSecret != nil && api_token != "" {
+ returnHash := fmt.Sprintf("%s+%d", hash, req.ContentLength)
+ apiToken := GetApiToken(req)
+ if PermissionSecret != nil && apiToken != "" {
expiry := time.Now().Add(blob_signature_ttl)
- return_hash = SignLocator(return_hash, api_token, expiry)
+ returnHash = SignLocator(returnHash, apiToken, expiry)
}
- resp.Write([]byte(return_hash + "\n"))
+ resp.Write([]byte(returnHash + "\n"))
}
-// IndexHandler
-// A HandleFunc to address /index and /index/{prefix} requests.
-//
+// IndexHandler is a HandleFunc to address /index and /index/{prefix} requests.
func IndexHandler(resp http.ResponseWriter, req *http.Request) {
// Reject unauthorized requests.
if !IsDataManagerToken(GetApiToken(req)) {
// * bytes_free
// * bytes_used
+// PoolStatus struct
type PoolStatus struct {
Alloc uint64 `json:"BytesAllocated"`
Cap int `json:"BuffersMax"`
Len int `json:"BuffersInUse"`
}
+// NodeStatus struct
type NodeStatus struct {
Volumes []*VolumeStatus `json:"volumes"`
BufferPool PoolStatus
var st NodeStatus
var stLock sync.Mutex
+// StatusHandler addresses /status.json requests.
func StatusHandler(resp http.ResponseWriter, req *http.Request) {
stLock.Lock()
readNodeStatus(&st)
If the JSON unmarshalling fails, return 400 Bad Request.
*/
+// PullRequest consists of a block locator and an ordered list of servers
type PullRequest struct {
Locator string `json:"locator"`
Servers []string `json:"servers"`
}
+// PullHandler processes "PUT /pull" requests for the data manager.
func PullHandler(resp http.ResponseWriter, req *http.Request) {
// Reject unauthorized requests.
if !IsDataManagerToken(GetApiToken(req)) {
pullq.ReplaceQueue(plist)
}
+// TrashRequest consists of a block locator and it's Mtime
type TrashRequest struct {
Locator string `json:"locator"`
BlockMtime int64 `json:"block_mtime"`
}
+// TrashHandler processes /trash requests.
func TrashHandler(resp http.ResponseWriter, req *http.Request) {
// Reject unauthorized requests.
if !IsDataManagerToken(GetApiToken(req)) {
// should be the only part of the code that cares about which volume a
// block is stored on, so it should be responsible for figuring out
// which volume to check for fetching blocks, storing blocks, etc.
-
// ==============================
+
// GetBlock fetches and returns the block identified by "hash".
//
// On success, GetBlock returns a byte slice with the block data, and
// If the block found does not have the correct MD5 hash, returns
// DiskHashError.
//
-
func GetBlock(hash string) ([]byte, error) {
// Attempt to read the requested hash from a keep volume.
- error_to_caller := NotFoundError
+ errorToCaller := NotFoundError
for _, vol := range KeepVM.AllReadable() {
buf, err := vol.Get(hash)
// this.
log.Printf("%s: checksum mismatch for request %s (actual %s)",
vol, hash, filehash)
- error_to_caller = DiskHashError
+ errorToCaller = DiskHashError
bufs.Put(buf)
continue
}
- if error_to_caller == DiskHashError {
+ if errorToCaller == DiskHashError {
log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned",
vol, hash)
}
return buf, nil
}
- return nil, error_to_caller
+ return nil, errorToCaller
}
-/* PutBlock(block, hash)
- Stores the BLOCK (identified by the content id HASH) in Keep.
-
- The MD5 checksum of the block must be identical to the content id HASH.
- If not, an error is returned.
-
- PutBlock stores the BLOCK on the first Keep volume with free space.
- A failure code is returned to the user only if all volumes fail.
-
- On success, PutBlock returns nil.
- On failure, it returns a KeepError with one of the following codes:
-
- 500 Collision
- A different block with the same hash already exists on this
- Keep server.
- 422 MD5Fail
- The MD5 hash of the BLOCK does not match the argument HASH.
- 503 Full
- There was not enough space left in any Keep volume to store
- the object.
- 500 Fail
- The object could not be stored for some other reason (e.g.
- all writes failed). The text of the error message should
- provide as much detail as possible.
-*/
-
+// PutBlock Stores the BLOCK (identified by the content id HASH) in Keep.
+//
+// PutBlock(block, hash)
+// Stores the BLOCK (identified by the content id HASH) in Keep.
+//
+// The MD5 checksum of the block must be identical to the content id HASH.
+// If not, an error is returned.
+//
+// PutBlock stores the BLOCK on the first Keep volume with free space.
+// A failure code is returned to the user only if all volumes fail.
+//
+// On success, PutBlock returns nil.
+// On failure, it returns a KeepError with one of the following codes:
+//
+// 500 Collision
+// A different block with the same hash already exists on this
+// Keep server.
+// 422 MD5Fail
+// The MD5 hash of the BLOCK does not match the argument HASH.
+// 503 Full
+// There was not enough space left in any Keep volume to store
+// the object.
+// 500 Fail
+// The object could not be stored for some other reason (e.g.
+// all writes failed). The text of the error message should
+// provide as much detail as possible.
+//
func PutBlock(block []byte, hash string) error {
// Check that BLOCK's checksum matches HASH.
blockhash := fmt.Sprintf("%x", md5.Sum(block))
if allFull {
log.Print("All volumes are full.")
return FullError
- } else {
- // Already logged the non-full errors.
- return GenericError
}
+ // Already logged the non-full errors.
+ return GenericError
}
// CompareAndTouch returns nil if one of the volumes already has the
var validLocatorRe = regexp.MustCompile(`^[0-9a-f]{32}$`)
-// IsValidLocator
-// Return true if the specified string is a valid Keep locator.
-// When Keep is extended to support hash types other than MD5,
-// this should be updated to cover those as well.
+// IsValidLocator returns true if the specified string is a valid Keep locator.
+// When Keep is extended to support hash types other than MD5,
+// this should be updated to cover those as well.
//
func IsValidLocator(loc string) bool {
return validLocatorRe.MatchString(loc)
var authRe = regexp.MustCompile(`^OAuth2\s+(.*)`)
-// GetApiToken returns the OAuth2 token from the Authorization
+// GetAPIToken returns the OAuth2 token from the Authorization
// header of a HTTP request, or an empty string if no matching
// token is found.
func GetApiToken(req *http.Request) string {
}
// IsExpired returns true if the given Unix timestamp (expressed as a
-// hexadecimal string) is in the past, or if timestamp_hex cannot be
+// hexadecimal string) is in the past, or if timestampHex cannot be
// parsed as a hexadecimal string.
-func IsExpired(timestamp_hex string) bool {
- ts, err := strconv.ParseInt(timestamp_hex, 16, 0)
+func IsExpired(timestampHex string) bool {
+ ts, err := strconv.ParseInt(timestampHex, 16, 0)
if err != nil {
log.Printf("IsExpired: %s", err)
return true
return time.Unix(ts, 0).Before(time.Now())
}
-// CanDelete returns true if the user identified by api_token is
+// CanDelete returns true if the user identified by apiToken is
// allowed to delete blocks.
-func CanDelete(api_token string) bool {
- if api_token == "" {
+func CanDelete(apiToken string) bool {
+ if apiToken == "" {
return false
}
// Blocks may be deleted only when Keep has been configured with a
// data manager.
- if IsDataManagerToken(api_token) {
+ if IsDataManagerToken(apiToken) {
return true
}
- // TODO(twp): look up api_token with the API server
+ // TODO(twp): look up apiToken with the API server
// return true if is_admin is true and if the token
// has unlimited scope
return false
}
-// IsDataManagerToken returns true if api_token represents the data
+// IsDataManagerToken returns true if apiToken represents the data
// manager's token.
-func IsDataManagerToken(api_token string) bool {
- return data_manager_token != "" && api_token == data_manager_token
+func IsDataManagerToken(apiToken string) bool {
+ return data_manager_token != "" && apiToken == data_manager_token
}
// Default TCP address on which to listen for requests.
// Initialized by the --listen flag.
-const DEFAULT_ADDR = ":25107"
+const DefaultAddr = ":25107"
// A Keep "block" is 64MB.
-const BLOCKSIZE = 64 * 1024 * 1024
+const BlockSize = 64 * 1024 * 1024
-// A Keep volume must have at least MIN_FREE_KILOBYTES available
+// A Keep volume must have at least MinFreeKilobytes available
// in order to permit writes.
-const MIN_FREE_KILOBYTES = BLOCKSIZE / 1024
+const MinFreeKilobytes = BlockSize / 1024
-var PROC_MOUNTS = "/proc/mounts"
+// ProcMounts /proc/mounts
+var ProcMounts = "/proc/mounts"
-// enforce_permissions controls whether permission signatures
+// enforcePermissions controls whether permission signatures
// should be enforced (affecting GET and DELETE requests).
// Initialized by the -enforce-permissions flag.
-var enforce_permissions bool
+var enforcePermissions bool
// blob_signature_ttl is the time duration for which new permission
// signatures (returned by PUT requests) will be valid.
var maxBuffers = 128
var bufs *bufferPool
-// ==========
-// Error types.
+// KeepError types.
//
type KeepError struct {
HTTPCode int
// other than "/". It returns the number of volumes added.
func (vs *volumeSet) Discover() int {
added := 0
- f, err := os.Open(PROC_MOUNTS)
+ f, err := os.Open(ProcMounts)
if err != nil {
- log.Fatalf("opening %s: %s", PROC_MOUNTS, err)
+ log.Fatalf("opening %s: %s", ProcMounts, err)
}
scanner := bufio.NewScanner(f)
for scanner.Scan() {
args := strings.Fields(scanner.Text())
if err := scanner.Err(); err != nil {
- log.Fatalf("reading %s: %s", PROC_MOUNTS, err)
+ log.Fatalf("reading %s: %s", ProcMounts, err)
}
dev, mount := args[0], args[1]
if mount == "/" {
"File with the API token used by the Data Manager. All DELETE "+
"requests or GET /index requests must carry this token.")
flag.BoolVar(
- &enforce_permissions,
+ &enforcePermissions,
"enforce-permissions",
false,
"Enforce permission signatures on requests.")
flag.StringVar(
&listen,
"listen",
- DEFAULT_ADDR,
+ DefaultAddr,
"Listening address, in the form \"host:port\". e.g., 10.0.1.24:8000. Omit the host part to listen on all interfaces.")
flag.BoolVar(
&never_delete,
&maxBuffers,
"max-buffers",
maxBuffers,
- fmt.Sprintf("Maximum RAM to use for data buffers, given in multiples of block size (%d MiB). When this limit is reached, HTTP requests requiring buffers (like GET and PUT) will wait for buffer space to be released.", BLOCKSIZE>>20))
+ fmt.Sprintf("Maximum RAM to use for data buffers, given in multiples of block size (%d MiB). When this limit is reached, HTTP requests requiring buffers (like GET and PUT) will wait for buffer space to be released.", BlockSize>>20))
flag.Parse()
if maxBuffers < 0 {
log.Fatal("-max-buffers must be greater than zero.")
}
- bufs = newBufferPool(maxBuffers, BLOCKSIZE)
+ bufs = newBufferPool(maxBuffers, BlockSize)
if pidfile != "" {
f, err := os.OpenFile(pidfile, os.O_RDWR|os.O_CREATE, 0777)
blob_signature_ttl = time.Duration(permission_ttl_sec) * time.Second
if PermissionSecret == nil {
- if enforce_permissions {
+ if enforcePermissions {
log.Fatal("-enforce-permissions requires a permission key")
} else {
log.Println("Running without a PermissionSecret. Block locators " +
projects / arvados.git / commitdiff