sdk/go/manifest/manifest.go

   1 /* Deals with parsing Manifest Text. */
   2
   3 // Inspired by the Manifest class in arvados/sdk/ruby/lib/arvados/keep.rb
   4
   5 package manifest
   6
   7 import (
   8         "errors"
   9         "fmt"
  10         "git.curoverse.com/arvados.git/sdk/go/blockdigest"
  11         "path"
  12         "regexp"
  13         "sort"
  14         "strconv"
  15         "strings"
  16 )
  17
  18 var ErrInvalidToken = errors.New("Invalid token")
  19
  20 type Manifest struct {
  21         Text string
  22         Err  error
  23 }
  24
  25 type BlockLocator struct {
  26         Digest blockdigest.BlockDigest
  27         Size   int
  28         Hints  []string
  29 }
  30
  31 // FileSegment is a portion of a file that is contained within a
  32 // single block.
  33 type FileSegment struct {
  34         Locator string
  35         // Offset (within this block) of this data segment
  36         Offset int
  37         Len    int
  38 }
  39
  40 // FileStreamSegment is a portion of a file described as a segment of a stream.
  41 type FileStreamSegment struct {
  42         SegPos uint64
  43         SegLen uint64
  44         Name   string
  45 }
  46
  47 // Represents a single line from a manifest.
  48 type ManifestStream struct {
  49         StreamName         string
  50         Blocks             []string
  51         blockOffsets       []uint64
  52         FileStreamSegments []FileStreamSegment
  53         Err                error
  54 }
  55
  56 // Array of segments referencing file content
  57 type segmentedFile []FileSegment
  58
  59 // Map of files to list of file segments referencing file content
  60 type segmentedStream map[string]segmentedFile
  61
  62 // Map of streams
  63 type segmentedManifest map[string]segmentedStream
  64
  65 var escapeSeq = regexp.MustCompile(`\\([0-9]{3}|\\)`)
  66
  67 func unescapeSeq(seq string) string {
  68         if seq == `\\` {
  69                 return `\`
  70         }
  71         i, err := strconv.ParseUint(seq[1:], 8, 8)
  72         if err != nil {
  73                 // Invalid escape sequence: can't unescape.
  74                 return seq
  75         }
  76         return string([]byte{byte(i)})
  77 }
  78
  79 func EscapeName(s string) string {
  80         raw := []byte(s)
  81         escaped := make([]byte, 0, len(s))
  82         for _, c := range raw {
  83                 if c <= 32 {
  84                         oct := fmt.Sprintf("\\%03o", c)
  85                         escaped = append(escaped, []byte(oct)...)
  86                 } else {
  87                         escaped = append(escaped, c)
  88                 }
  89         }
  90         return string(escaped)
  91 }
  92
  93 func UnescapeName(s string) string {
  94         return escapeSeq.ReplaceAllStringFunc(s, unescapeSeq)
  95 }
  96
  97 func ParseBlockLocator(s string) (b BlockLocator, err error) {
  98         if !blockdigest.LocatorPattern.MatchString(s) {
  99                 err = fmt.Errorf("String \"%s\" does not match BlockLocator pattern "+
 100                         "\"%s\".",
 101                         s,
 102                         blockdigest.LocatorPattern.String())
 103         } else {
 104                 tokens := strings.Split(s, "+")
 105                 var blockSize int64
 106                 var blockDigest blockdigest.BlockDigest
 107                 // We expect both of the following to succeed since LocatorPattern
 108                 // restricts the strings appropriately.
 109                 blockDigest, err = blockdigest.FromString(tokens[0])
 110                 if err != nil {
 111                         return
 112                 }
 113                 blockSize, err = strconv.ParseInt(tokens[1], 10, 0)
 114                 if err != nil {
 115                         return
 116                 }
 117                 b.Digest = blockDigest
 118                 b.Size = int(blockSize)
 119                 b.Hints = tokens[2:]
 120         }
 121         return
 122 }
 123
 124 func parseFileStreamSegment(tok string) (ft FileStreamSegment, err error) {
 125         parts := strings.SplitN(tok, ":", 3)
 126         if len(parts) != 3 {
 127                 err = ErrInvalidToken
 128                 return
 129         }
 130         ft.SegPos, err = strconv.ParseUint(parts[0], 10, 64)
 131         if err != nil {
 132                 return
 133         }
 134         ft.SegLen, err = strconv.ParseUint(parts[1], 10, 64)
 135         if err != nil {
 136                 return
 137         }
 138         ft.Name = UnescapeName(parts[2])
 139         return
 140 }
 141
 142 func (s *ManifestStream) FileSegmentIterByName(filepath string) <-chan *FileSegment {
 143         ch := make(chan *FileSegment, 64)
 144         go func() {
 145                 s.sendFileSegmentIterByName(filepath, ch)
 146                 close(ch)
 147         }()
 148         return ch
 149 }
 150
 151 func firstBlock(offsets []uint64, range_start uint64) int {
 152         // range_start/block_start is the inclusive lower bound
 153         // range_end/block_end is the exclusive upper bound
 154
 155         hi := len(offsets) - 1
 156         var lo int
 157         i := ((hi + lo) / 2)
 158         block_start := offsets[i]
 159         block_end := offsets[i+1]
 160
 161         // perform a binary search for the first block
 162         // assumes that all of the blocks are contiguous, so range_start is guaranteed
 163         // to either fall into the range of a block or be outside the block range entirely
 164         for !(range_start >= block_start && range_start < block_end) {
 165                 fmt.Println(i, block_start, block_end)
 166                 if lo == i {
 167                         // must be out of range, fail
 168                         return -1
 169                 }
 170                 if range_start > block_start {
 171                         lo = i
 172                 } else {
 173                         hi = i
 174                 }
 175                 i = ((hi + lo) / 2)
 176                 block_start = offsets[i]
 177                 block_end = offsets[i+1]
 178         }
 179         return i
 180 }
 181
 182 func (s *ManifestStream) sendFileSegmentIterByName(filepath string, ch chan<- *FileSegment) {
 183         // This is what streamName+"/"+fileName will look like:
 184         target := fixStreamName(filepath)
 185         for _, fTok := range s.FileStreamSegments {
 186                 wantPos := fTok.SegPos
 187                 wantLen := fTok.SegLen
 188                 name := fTok.Name
 189
 190                 if s.StreamName+"/"+name != target {
 191                         continue
 192                 }
 193                 if wantLen == 0 {
 194                         ch <- &FileSegment{Locator: "d41d8cd98f00b204e9800998ecf8427e+0", Offset: 0, Len: 0}
 195                         continue
 196                 }
 197
 198                 // Binary search to determine first block in the stream
 199                 i := firstBlock(s.blockOffsets, wantPos)
 200                 if i == -1 {
 201                         // Shouldn't happen, file segments are checked in parseManifestStream
 202                         panic(fmt.Sprintf("File segment %v extends past end of stream", fTok))
 203                 }
 204                 for ; i < len(s.Blocks); i++ {
 205                         blockPos := s.blockOffsets[i]
 206                         blockEnd := s.blockOffsets[i+1]
 207                         if blockEnd <= wantPos {
 208                                 // Shouldn't happen, FirstBlock() should start
 209                                 // us on the right block, so if this triggers
 210                                 // that means there is a bug.
 211                                 panic(fmt.Sprintf("Block end %v comes before start of file segment %v", blockEnd, wantPos))
 212                         }
 213                         if blockPos >= wantPos+wantLen {
 214                                 // current block comes after current file span
 215                                 break
 216                         }
 217
 218                         fseg := FileSegment{
 219                                 Locator: s.Blocks[i],
 220                                 Offset:  0,
 221                                 Len:     int(blockEnd - blockPos),
 222                         }
 223                         if blockPos < wantPos {
 224                                 fseg.Offset = int(wantPos - blockPos)
 225                                 fseg.Len -= fseg.Offset
 226                         }
 227                         if blockEnd > wantPos+wantLen {
 228                                 fseg.Len = int(wantPos+wantLen-blockPos) - fseg.Offset
 229                         }
 230                         ch <- &fseg
 231                 }
 232         }
 233 }
 234
 235 func parseManifestStream(s string) (m ManifestStream) {
 236         tokens := strings.Split(s, " ")
 237
 238         m.StreamName = UnescapeName(tokens[0])
 239         if m.StreamName != "." && !strings.HasPrefix(m.StreamName, "./") {
 240                 m.Err = fmt.Errorf("Invalid stream name: %s", m.StreamName)
 241                 return
 242         }
 243
 244         tokens = tokens[1:]
 245         var i int
 246         for i = 0; i < len(tokens); i++ {
 247                 if !blockdigest.IsBlockLocator(tokens[i]) {
 248                         break
 249                 }
 250         }
 251         m.Blocks = tokens[:i]
 252         fileTokens := tokens[i:]
 253
 254         if len(m.Blocks) == 0 {
 255                 m.Err = fmt.Errorf("No block locators found")
 256                 return
 257         }
 258
 259         m.blockOffsets = make([]uint64, len(m.Blocks)+1)
 260         var streamoffset uint64
 261         for i, b := range m.Blocks {
 262                 bl, err := ParseBlockLocator(b)
 263                 if err != nil {
 264                         m.Err = err
 265                         return
 266                 }
 267                 m.blockOffsets[i] = streamoffset
 268                 streamoffset += uint64(bl.Size)
 269         }
 270         m.blockOffsets[len(m.Blocks)] = streamoffset
 271
 272         if len(fileTokens) == 0 {
 273                 m.Err = fmt.Errorf("No file tokens found")
 274                 return
 275         }
 276
 277         for _, ft := range fileTokens {
 278                 pft, err := parseFileStreamSegment(ft)
 279                 if err != nil {
 280                         m.Err = fmt.Errorf("Invalid file token: %s", ft)
 281                         break
 282                 }
 283                 if pft.SegPos+pft.SegLen > streamoffset {
 284                         m.Err = fmt.Errorf("File segment %s extends past end of stream %d", ft, streamoffset)
 285                         break
 286                 }
 287                 m.FileStreamSegments = append(m.FileStreamSegments, pft)
 288         }
 289
 290         return
 291 }
 292
 293 func fixStreamName(sn string) string {
 294         sn = path.Clean(sn)
 295         if strings.HasPrefix(sn, "/") {
 296                 sn = "." + sn
 297         } else if sn != "." {
 298                 sn = "./" + sn
 299         }
 300         return sn
 301 }
 302
 303 func splitPath(srcpath string) (streamname, filename string) {
 304         pathIdx := strings.LastIndex(srcpath, "/")
 305         if pathIdx >= 0 {
 306                 streamname = srcpath[0:pathIdx]
 307                 filename = srcpath[pathIdx+1:]
 308         } else {
 309                 streamname = srcpath
 310                 filename = ""
 311         }
 312         return
 313 }
 314
 315 func (m *Manifest) segment() (*segmentedManifest, error) {
 316         files := make(segmentedManifest)
 317
 318         for stream := range m.StreamIter() {
 319                 if stream.Err != nil {
 320                         // Stream has an error
 321                         return nil, stream.Err
 322                 }
 323                 currentStreamfiles := make(map[string]bool)
 324                 for _, f := range stream.FileStreamSegments {
 325                         sn := stream.StreamName
 326                         if strings.HasSuffix(sn, "/") {
 327                                 sn = sn[0 : len(sn)-1]
 328                         }
 329                         path := sn + "/" + f.Name
 330                         streamname, filename := splitPath(path)
 331                         if files[streamname] == nil {
 332                                 files[streamname] = make(segmentedStream)
 333                         }
 334                         if !currentStreamfiles[path] {
 335                                 segs := files[streamname][filename]
 336                                 for seg := range stream.FileSegmentIterByName(path) {
 337                                         if seg.Len > 0 {
 338                                                 segs = append(segs, *seg)
 339                                         }
 340                                 }
 341                                 files[streamname][filename] = segs
 342                                 currentStreamfiles[path] = true
 343                         }
 344                 }
 345         }
 346
 347         return &files, nil
 348 }
 349
 350 func (stream segmentedStream) normalizedText(name string) string {
 351         var sortedfiles []string
 352         for k, _ := range stream {
 353                 sortedfiles = append(sortedfiles, k)
 354         }
 355         sort.Strings(sortedfiles)
 356
 357         stream_tokens := []string{EscapeName(name)}
 358
 359         blocks := make(map[string]int64)
 360         var streamoffset int64
 361
 362         // Go through each file and add each referenced block exactly once.
 363         for _, streamfile := range sortedfiles {
 364                 for _, segment := range stream[streamfile] {
 365                         if _, ok := blocks[segment.Locator]; !ok {
 366                                 stream_tokens = append(stream_tokens, segment.Locator)
 367                                 blocks[segment.Locator] = streamoffset
 368                                 b, _ := ParseBlockLocator(segment.Locator)
 369                                 streamoffset += int64(b.Size)
 370                         }
 371                 }
 372         }
 373
 374         if len(stream_tokens) == 1 {
 375                 stream_tokens = append(stream_tokens, "d41d8cd98f00b204e9800998ecf8427e+0")
 376         }
 377
 378         for _, streamfile := range sortedfiles {
 379                 // Add in file segments
 380                 span_start := int64(-1)
 381                 span_end := int64(0)
 382                 fout := EscapeName(streamfile)
 383                 for _, segment := range stream[streamfile] {
 384                         // Collapse adjacent segments
 385                         streamoffset = blocks[segment.Locator] + int64(segment.Offset)
 386                         if span_start == -1 {
 387                                 span_start = streamoffset
 388                                 span_end = streamoffset + int64(segment.Len)
 389                         } else {
 390                                 if streamoffset == span_end {
 391                                         span_end += int64(segment.Len)
 392                                 } else {
 393                                         stream_tokens = append(stream_tokens, fmt.Sprintf("%d:%d:%s", span_start, span_end-span_start, fout))
 394                                         span_start = streamoffset
 395                                         span_end = streamoffset + int64(segment.Len)
 396                                 }
 397                         }
 398                 }
 399
 400                 if span_start != -1 {
 401                         stream_tokens = append(stream_tokens, fmt.Sprintf("%d:%d:%s", span_start, span_end-span_start, fout))
 402                 }
 403
 404                 if len(stream[streamfile]) == 0 {
 405                         stream_tokens = append(stream_tokens, fmt.Sprintf("0:0:%s", fout))
 406                 }
 407         }
 408
 409         return strings.Join(stream_tokens, " ") + "\n"
 410 }
 411
 412 func (m segmentedManifest) manifestTextForPath(srcpath, relocate string) string {
 413         srcpath = fixStreamName(srcpath)
 414
 415         var suffix string
 416         if strings.HasSuffix(relocate, "/") {
 417                 suffix = "/"
 418         }
 419         relocate = fixStreamName(relocate) + suffix
 420
 421         streamname, filename := splitPath(srcpath)
 422
 423         if stream, ok := m[streamname]; ok {
 424                 // check if it refers to a single file in a stream
 425                 filesegs, okfile := stream[filename]
 426                 if okfile {
 427                         newstream := make(segmentedStream)
 428                         relocate_stream, relocate_filename := splitPath(relocate)
 429                         if relocate_filename == "" {
 430                                 relocate_filename = filename
 431                         }
 432                         newstream[relocate_filename] = filesegs
 433                         return newstream.normalizedText(relocate_stream)
 434                 }
 435         }
 436
 437         // Going to extract multiple streams
 438         prefix := srcpath + "/"
 439
 440         if strings.HasSuffix(relocate, "/") {
 441                 relocate = relocate[0 : len(relocate)-1]
 442         }
 443
 444         var sortedstreams []string
 445         for k, _ := range m {
 446                 sortedstreams = append(sortedstreams, k)
 447         }
 448         sort.Strings(sortedstreams)
 449
 450         manifest := ""
 451         for _, k := range sortedstreams {
 452                 if strings.HasPrefix(k, prefix) || k == srcpath {
 453                         manifest += m[k].normalizedText(relocate + k[len(srcpath):])
 454                 }
 455         }
 456         return manifest
 457 }
 458
 459 // Extract extracts some or all of the manifest and returns the extracted
 460 // portion as a normalized manifest.  This is a swiss army knife function that
 461 // can be several ways:
 462 //
 463 // If 'srcpath' and 'relocate' are '.' it simply returns an equivalent manifest
 464 // in normalized form.
 465 //
 466 // If 'srcpath' points to a single file, it will return manifest text for just that file.
 467 // The value of "relocate" is can be used to rename the file or set the file stream.
 468 //
 469 // ManifestTextForPath("./foo", ".")  (extract file "foo" and put it in stream ".")
 470 // ManifestTextForPath("./foo", "./bar")  (extract file "foo", rename it to "bar" in stream ".")
 471 // ManifestTextForPath("./foo", "./bar/") (extract file "foo", rename it to "./bar/foo")
 472 // ManifestTextForPath("./foo", "./bar/baz") (extract file "foo", rename it to "./bar/baz")
 473 //
 474 // Otherwise it will return the manifest text for all streams with the prefix in "srcpath" and place
 475 // them under the path in "relocate".
 476 //
 477 // ManifestTextForPath(".", ".")  (return entire normalized manfest text)
 478 // ManifestTextForPath("./stream", ".")  (extract "./stream" to "." and "./stream/subdir" to "./subdir")
 479 // ManifestTextForPath("./stream", "./bar")  (extract "./stream" to "./bar" and "./stream/subdir" to "./bar/subdir")
 480 func (m Manifest) Extract(srcpath, relocate string) (ret Manifest) {
 481         segmented, err := m.segment()
 482         if err != nil {
 483                 ret.Err = err
 484                 return
 485         }
 486         ret.Text = segmented.manifestTextForPath(srcpath, relocate)
 487         return
 488 }
 489
 490 func (m *Manifest) StreamIter() <-chan ManifestStream {
 491         ch := make(chan ManifestStream)
 492         go func(input string) {
 493                 // This slice holds the current line and the remainder of the
 494                 // manifest.  We parse one line at a time, to save effort if we
 495                 // only need the first few lines.
 496                 lines := []string{"", input}
 497                 for {
 498                         lines = strings.SplitN(lines[1], "\n", 2)
 499                         if len(lines[0]) > 0 {
 500                                 // Only parse non-blank lines
 501                                 ch <- parseManifestStream(lines[0])
 502                         }
 503                         if len(lines) == 1 {
 504                                 break
 505                         }
 506                 }
 507                 close(ch)
 508         }(m.Text)
 509         return ch
 510 }
 511
 512 func (m *Manifest) FileSegmentIterByName(filepath string) <-chan *FileSegment {
 513         ch := make(chan *FileSegment, 64)
 514         filepath = fixStreamName(filepath)
 515         go func() {
 516                 for stream := range m.StreamIter() {
 517                         if !strings.HasPrefix(filepath, stream.StreamName+"/") {
 518                                 continue
 519                         }
 520                         stream.sendFileSegmentIterByName(filepath, ch)
 521                 }
 522                 close(ch)
 523         }()
 524         return ch
 525 }
 526
 527 // Blocks may appear multiple times within the same manifest if they
 528 // are used by multiple files. In that case this Iterator will output
 529 // the same block multiple times.
 530 //
 531 // In order to detect parse errors, caller must check m.Err after the returned channel closes.
 532 func (m *Manifest) BlockIterWithDuplicates() <-chan blockdigest.BlockLocator {
 533         blockChannel := make(chan blockdigest.BlockLocator)
 534         go func(streamChannel <-chan ManifestStream) {
 535                 for ms := range streamChannel {
 536                         if ms.Err != nil {
 537                                 m.Err = ms.Err
 538                                 continue
 539                         }
 540                         for _, block := range ms.Blocks {
 541                                 if b, err := blockdigest.ParseBlockLocator(block); err == nil {
 542                                         blockChannel <- b
 543                                 } else {
 544                                         m.Err = err
 545                                 }
 546                         }
 547                 }
 548                 close(blockChannel)
 549         }(m.StreamIter())
 550         return blockChannel
 551 }