func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
	span := c.tracer.StartSpan("Client.ExperimentalReplayImport")
	defer span.Finish()

	// make work channel
	work := make(chan *importLog, concurrency*2)

	// spawn <concurrency> workers to read from channel
	eg := &errgroup.Group{}
	for i := 0; i < concurrency; i++ {
		eg.Go(func() error {
			for log := range work {
				// regular import doesn't forward to replicas, so we have to get all
				// the nodes.
				nodes, err := c.fetchFragmentNodes(log.Index, log.Shard)
				if err != nil {
					return errors.Wrap(err, "fetching fragment nodes")
				}

				if !log.IsRoaring {
					for _, node := range nodes {
						resp, err := c.doRequest(node.URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
						if err = anyError(resp, err); err != nil {
							return errors.Wrap(err, "doing import")
						}
						resp.Body.Close()
					}
				} else {
					// import-roaring forwards on to all replicas, so we only import to
					// one node.
					resp, err := c.doRequest(nodes[0].URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
					if err = anyError(resp, err); err != nil {
						return errors.Wrap(err, "doing import")
					}
					resp.Body.Close()
				}
			}
			return nil
		})
	}

	// wait on importers asynchronously in case they return early with errors
	waitErrCh := make(chan error)
	go func() {
		waitErrCh <- eg.Wait()
		close(waitErrCh)
	}()

	// populate work channel
	dec := newImportLogDecoder(r)
	var err error
	for {
		log := importLog{}
		err = dec.Decode(&log)
		if err != nil {
			err = errors.Wrap(err, "decoding")
			break
		}
		select {
		case work <- &log:
			continue
		case err = <-waitErrCh:
			err = errors.Wrap(err, "waiting")
		}
		break
	}

	// close work channel (now workers can exit)
	close(work)
	<-waitErrCh // make sure workers are complete
	if errors.Cause(err) == io.EOF {
		return nil
	}
	return err
}
func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
	span := c.tracer.StartSpan("Client.ExperimentalReplayImport")
	defer span.Finish()

	
	work := make(chan *importLog, concurrency*2)

	
	eg := &errgroup.Group{}
	for i := 0; i < concurrency; i++ {
		eg.Go(func() error {
			for log := range work {
				
				
				nodes, err := c.fetchFragmentNodes(log.Index, log.Shard)
				if err != nil {
					return errors.Wrap(err, "fetching fragment nodes")
				}

				if !log.IsRoaring {
					for _, node := range nodes {
						resp, err := c.doRequest(node.URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
						if err = anyError(resp, err); err != nil {
							return errors.Wrap(err, "doing import")
						}
						resp.Body.Close()
					}
				} else {
					
					
					resp, err := c.doRequest(nodes[0].URI(), "POST", log.Path, defaultProtobufHeaders(), bytes.NewReader(log.Data))
					if err = anyError(resp, err); err != nil {
						return errors.Wrap(err, "doing import")
					}
					resp.Body.Close()
				}
			}
			return nil
		})
	}

	
	waitErrCh := make(chan error)
	go func() {
		waitErrCh <- eg.Wait()
		close(waitErrCh)
	}()

	
	dec := newImportLogDecoder(r)
	var err error
	for {
		log := importLog{}
		err = dec.Decode(&log)
		if err != nil {
			err = errors.Wrap(err, "decoding")
			break
		}
		select {
		case work <- &log:
			continue
		case err = <-waitErrCh:
			err = errors.Wrap(err, "waiting")
		}
		break
	}

	
	close(work)
	<-waitErrCh 
	if errors.Cause(err) == io.EOF {
		return nil
	}
	return err
}
func ( c * Client ) ExperimentalReplayImport ( r io . Reader , concurrency int ) error { span := c . tracer . StartSpan ( " " ) defer span . Finish ( ) // make work channel work := make ( chan * importLog , concurrency * 2 ) // spawn <concurrency> workers to read from channel eg := & errgroup . Group { } for i := 0 ; i < concurrency ; i ++ { eg . Go ( func ( ) error { for log := range work { // regular import doesn't forward to replicas, so we have to get all // the nodes. nodes , err := c . fetchFragmentNodes ( log . Index , log . Shard ) if err != nil { return errors . Wrap ( err , " " ) } if ! log . IsRoaring { for _ , node := range nodes { resp , err := c . doRequest ( node . URI ( ) , " " , log . Path , defaultProtobufHeaders ( ) , bytes . NewReader ( log . Data ) ) if err = anyError ( resp , err ) ; err != nil { return errors . Wrap ( err , " " ) } resp . Body . Close ( ) } } else { // import-roaring forwards on to all replicas, so we only import to // one node. resp , err := c . doRequest ( nodes [ 0 ] . URI ( ) , " " , log . Path , defaultProtobufHeaders ( ) , bytes . NewReader ( log . Data ) ) if err = anyError ( resp , err ) ; err != nil { return errors . Wrap ( err , " " ) } resp . Body . Close ( ) } } return nil } ) } // wait on importers asynchronously in case they return early with errors waitErrCh := make ( chan error ) go func ( ) { waitErrCh <- eg . Wait ( ) close ( waitErrCh ) } ( ) // populate work channel dec := newImportLogDecoder ( r ) var err error for { log := importLog { } err = dec . Decode ( & log ) if err != nil { err = errors . Wrap ( err , " " ) break } select { case work <- & log : continue case err = <- waitErrCh : err = errors . Wrap ( err , " " ) } break } // close work channel (now workers can exit) close ( work ) <- waitErrCh // make sure workers are complete if errors . Cause ( err ) == io . EOF { return nil } return err }
// ExperimentalReplayImport takes a data stream which was previously recorded by the import
// logging functionality and imports it to Pilosa. The target cluster need not
// be of the same size as the original cluster, but it must already have the
// necessary schema in place. It is an experimental method and may be changed or
// removed.
ExperimentalReplayImport takes a data stream which was previously recorded by the import logging functionality and imports it to Pilosa . The target cluster need not be of the same size as the original cluster but it must already have the necessary schema in place . It is an experimental method and may be changed or removed .