@@ -29,69 +29,105 @@ var (
2929 errTransferNoPendingIngesters = errors .New ("no pending ingesters" )
3030)
3131
32- // TransferChunks receives all the chunks from another ingester.
33- func (i * Ingester ) TransferChunks (stream client.Ingester_TransferChunksServer ) error {
34- fromIngesterID := ""
35- seriesReceived := 0
36- xfer := func () error {
37- userStates := newUserStates (i .limiter , i .cfg , i .metrics )
32+ // returns source ingesterID, number of received series, added chunks and error
33+ func (i * Ingester ) fillUserStatesFromStream (userStates * userStates , stream client.Ingester_TransferChunksServer ) (fromIngesterID string , seriesReceived int , retErr error ) {
34+ chunksAdded := 0.0
3835
39- for {
40- wireSeries , err := stream .Recv ()
41- if err == io .EOF {
42- break
43- }
44- if err != nil {
45- return errors .Wrap (err , "TransferChunks: Recv" )
46- }
36+ defer func () {
37+ if retErr != nil {
38+ // Ensure the in memory chunks are updated to reflect the number of dropped chunks from the transfer
39+ i .metrics .memoryChunks .Sub (chunksAdded )
4740
48- // We can't send "extra" fields with a streaming call, so we repeat
49- // wireSeries.FromIngesterId and assume it is the same every time
50- // round this loop.
51- if fromIngesterID == "" {
52- fromIngesterID = wireSeries .FromIngesterId
53- level .Info (util .Logger ).Log ("msg" , "processing TransferChunks request" , "from_ingester" , fromIngesterID )
41+ // If an error occurs during the transfer and the user state is to be discarded,
42+ // ensure the metrics it exports reflect this.
43+ userStates .teardown ()
44+ }
45+ }()
5446
55- // Before transfer, make sure 'from' ingester is in correct state to call ClaimTokensFor later
56- err := i .checkFromIngesterIsInLeavingState (stream .Context (), fromIngesterID )
57- if err != nil {
58- return errors .Wrap (err , "TransferChunks: checkFromIngesterIsInLeavingState" )
59- }
60- }
61- descs , err := fromWireChunks (wireSeries .Chunks )
62- if err != nil {
63- return errors .Wrap (err , "TransferChunks: fromWireChunks" )
64- }
47+ for {
48+ wireSeries , err := stream .Recv ()
49+ if err == io .EOF {
50+ break
51+ }
52+ if err != nil {
53+ retErr = errors .Wrap (err , "TransferChunks: Recv" )
54+ return
55+ }
6556
66- state , fp , series , err := userStates .getOrCreateSeries (stream .Context (), wireSeries .UserId , wireSeries .Labels , nil )
67- if err != nil {
68- return errors .Wrapf (err , "TransferChunks: getOrCreateSeries: user %s series %s" , wireSeries .UserId , wireSeries .Labels )
69- }
70- prevNumChunks := len (series .chunkDescs )
57+ // We can't send "extra" fields with a streaming call, so we repeat
58+ // wireSeries.FromIngesterId and assume it is the same every time
59+ // round this loop.
60+ if fromIngesterID == "" {
61+ fromIngesterID = wireSeries .FromIngesterId
62+ level .Info (util .Logger ).Log ("msg" , "processing TransferChunks request" , "from_ingester" , fromIngesterID )
7163
72- err = series . setChunks ( descs )
73- state . fpLocker . Unlock ( fp ) // acquired in getOrCreateSeries
64+ // Before transfer, make sure 'from' ingester is in correct state to call ClaimTokensFor later
65+ err := i . checkFromIngesterIsInLeavingState ( stream . Context (), fromIngesterID )
7466 if err != nil {
75- return errors .Wrapf (err , "TransferChunks: setChunks: user %s series %s" , wireSeries .UserId , wireSeries .Labels )
67+ retErr = errors .Wrap (err , "TransferChunks: checkFromIngesterIsInLeavingState" )
68+ return
7669 }
77-
78- seriesReceived ++
79- i .metrics .memoryChunks .Add (float64 (len (series .chunkDescs ) - prevNumChunks ))
80- i .metrics .receivedChunks .Add (float64 (len (descs )))
70+ }
71+ descs , err := fromWireChunks (wireSeries .Chunks )
72+ if err != nil {
73+ retErr = errors .Wrap (err , "TransferChunks: fromWireChunks" )
74+ return
8175 }
8276
83- if seriesReceived == 0 {
84- level .Error (util .Logger ).Log ("msg" , "received TransferChunks request with no series" , "from_ingester" , fromIngesterID )
85- return fmt .Errorf ("TransferChunks: no series" )
77+ state , fp , series , err := userStates .getOrCreateSeries (stream .Context (), wireSeries .UserId , wireSeries .Labels , nil )
78+ if err != nil {
79+ retErr = errors .Wrapf (err , "TransferChunks: getOrCreateSeries: user %s series %s" , wireSeries .UserId , wireSeries .Labels )
80+ return
8681 }
82+ prevNumChunks := len (series .chunkDescs )
8783
88- if fromIngesterID == "" {
89- level .Error (util .Logger ).Log ("msg" , "received TransferChunks request with no ID from ingester" )
90- return fmt .Errorf ("no ingester id" )
84+ err = series .setChunks (descs )
85+ state .fpLocker .Unlock (fp ) // acquired in getOrCreateSeries
86+ if err != nil {
87+ retErr = errors .Wrapf (err , "TransferChunks: setChunks: user %s series %s" , wireSeries .UserId , wireSeries .Labels )
88+ return
9189 }
9290
93- if err := i .lifecycler .ClaimTokensFor (stream .Context (), fromIngesterID ); err != nil {
94- return errors .Wrap (err , "TransferChunks: ClaimTokensFor" )
91+ seriesReceived ++
92+ chunksDelta := float64 (len (series .chunkDescs ) - prevNumChunks )
93+ chunksAdded += chunksDelta
94+ i .metrics .memoryChunks .Add (chunksDelta )
95+ i .metrics .receivedChunks .Add (float64 (len (descs )))
96+ }
97+
98+ if seriesReceived == 0 {
99+ level .Error (util .Logger ).Log ("msg" , "received TransferChunks request with no series" , "from_ingester" , fromIngesterID )
100+ retErr = fmt .Errorf ("TransferChunks: no series" )
101+ return
102+ }
103+
104+ if fromIngesterID == "" {
105+ level .Error (util .Logger ).Log ("msg" , "received TransferChunks request with no ID from ingester" )
106+ retErr = fmt .Errorf ("no ingester id" )
107+ return
108+ }
109+
110+ if err := i .lifecycler .ClaimTokensFor (stream .Context (), fromIngesterID ); err != nil {
111+ retErr = errors .Wrap (err , "TransferChunks: ClaimTokensFor" )
112+ return
113+ }
114+
115+ return
116+ }
117+
118+ // TransferChunks receives all the chunks from another ingester.
119+ func (i * Ingester ) TransferChunks (stream client.Ingester_TransferChunksServer ) error {
120+ fromIngesterID := ""
121+ seriesReceived := 0
122+
123+ xfer := func () error {
124+ userStates := newUserStates (i .limiter , i .cfg , i .metrics )
125+
126+ var err error
127+ fromIngesterID , seriesReceived , err = i .fillUserStatesFromStream (userStates , stream )
128+
129+ if err != nil {
130+ return err
95131 }
96132
97133 i .userStatesMtx .Lock ()
0 commit comments