WriteBackup uses the request values to create a stream writer then hand off the data retrieval to stream.Orchestrate. The writer will create all the fd's needed to collect the data and later move to the target. Returns errors on failure, nil on success.
(ctx context.Context)
| 513 | // collect the data and later move to the target. |
| 514 | // Returns errors on failure, nil on success. |
| 515 | func (pr *BackupProcessor) WriteBackup(ctx context.Context) (*pb.BackupResponse, error) { |
| 516 | if err := ctx.Err(); err != nil { |
| 517 | return nil, err |
| 518 | } |
| 519 | uri, err := url.Parse(pr.Request.Destination) |
| 520 | if err != nil { |
| 521 | return nil, err |
| 522 | } |
| 523 | handler, err := NewUriHandler(uri, GetCredentialsFromRequest(pr.Request)) |
| 524 | if err != nil { |
| 525 | return nil, err |
| 526 | } |
| 527 | w, err := createBackupFile(handler, uri, pr.Request) |
| 528 | if err != nil { |
| 529 | return nil, err |
| 530 | } |
| 531 | glog.V(3).Infof("Backup manifest version: %d", pr.Request.SinceTs) |
| 532 | |
| 533 | eWriter, err := enc.GetWriter(x.WorkerConfig.EncryptionKey, w) |
| 534 | if err != nil { |
| 535 | return nil, err |
| 536 | } |
| 537 | |
| 538 | // Snappy is much faster than gzip compression, even with the BestSpeed |
| 539 | // gzip option. In fact, in my experiments, gzip compression caused the |
| 540 | // output speed to be ~30 MBps. Snappy can write at ~90 MBps, and overall |
| 541 | // the speed is similar to writing uncompressed data on disk. |
| 542 | // |
| 543 | // These are the times I saw: |
| 544 | // Without compression: 7m2s 33GB output. |
| 545 | // With snappy: 7m11s 9.5GB output. |
| 546 | // With snappy + S3: 7m54s 9.5GB output. |
| 547 | cWriter := s2.NewWriter(eWriter) |
| 548 | |
| 549 | stream := pr.DB.NewStreamAt(pr.Request.ReadTs) |
| 550 | stream.LogPrefix = "Dgraph.Backup" |
| 551 | // Ignore versions less than given sinceTs timestamp, or skip older versions of |
| 552 | // the given key by returning an empty list. |
| 553 | // Do not do this for schema and type keys. Those keys always have a |
| 554 | // version of one. They're handled separately. |
| 555 | stream.SinceTs = pr.Request.SinceTs |
| 556 | stream.Prefix = []byte{x.ByteData} |
| 557 | |
| 558 | var response pb.BackupResponse |
| 559 | stream.KeyToList = func(key []byte, itr *badger.Iterator) (*bpb.KVList, error) { |
| 560 | tl := pr.threads[itr.ThreadId] |
| 561 | tl.alloc = itr.Alloc |
| 562 | |
| 563 | bitr := itr |
| 564 | // Use the threadlocal iterator because "itr" has the sinceTs set and |
| 565 | // it will not be able to read all the data. |
| 566 | if tl.itr != nil { |
| 567 | bitr = tl.itr |
| 568 | bitr.Seek(key) |
| 569 | } |
| 570 | |
| 571 | kvList, dropOp, err := tl.toBackupList(key, bitr) |
| 572 | if err != nil { |
no test coverage detected